webrtc中的任务队列TaskQueue

在webrtc中，有一个任务队列TaskQueue，在视频编码模块中就是通过它是实现编码线程，这篇文章将讲解它的实现和应用场景。

TaskQueue

-任务队列 TaskQueue的用途是将任务放到到另外一个线程执行，与std::aysnc功能相似，但是它不能取得执行结果。任务队列是多线程编程下的基础设施，是一种线程间交互手段。

一个TaskQueue对象就代表了一个任务执行线程，在构造时产生线程，在析构时结束线程。

TaskQeueue采用Pimpl技术实现，它提供的是接口，内部的impl_对象封装了具体实现，如下类图，TaskQueue通过impl_持有一个TaskQueueBase类型的对象。

它的构造函数传入一个TaskQueueBase的实例。

explicit TaskQueue(std::unique_ptr<webrtc::TaskQueueBase,
                                     webrtc::TaskQueueDeleter> task_queue);
1
2

TaskQueue的接口

这里列出它的主要的四个接口: PostTask,PostDelayedTask,isCurrent,Current

• PostTask和PostDelayedTask

一个是将task放入TaskQueue立即执行，一个将task放入TaskQueue，指定delay后时间执行。

TaskQueue类中有两个类型的接口PostTask和PostDelayedTask，重载了两种不同类型的参数，一个是传入QueueTask对象，一个是传入Closure(Clousre是模板参数)，Closure使用更加方便，它可以传入一个lambda表达式。

• isCurrent和Current

isCurrent可以用于判断任务是否运行在TaskQueue对象所代表的线程上
Current 获取任务线程关联的TaskQueue对象

这两个方法基于thread_local类型的变量实现(即线程局部变量)，每个线程都有自己的局部变量，在线程内可见，线程间不可见。所以不需要对该类型的变量进行互斥。

首先定义了一个thread_local类型的TaskQueueBase指针

thread_local TaskQueueBase* current = nullptr;
1

在线程执行时，赋值为this

那么isCurrent的实现如下：

bool IsCurrent() const { return Current() == this; }
1

Current的实现如下：

TaskQueueBase* TaskQueueBase::Current() {
  return current;
}
1
2
3

TaskQueue对象就代表了一个线程，它可以在任意其它线程使用，在任务中调用IsCurrent()方法可以判断它是否在TaskQueue对象所代表的线程上执行。通过这种方式人为的将任务分发到指定的线程，而避免加锁互斥。

TaskQueueBase具体子类

前面提到 TaskQueue只是封装接口，TaskQueueBase才是真正的实现，但它是个抽象接口类，如下图

有4个具体的实现类``TaskQueueStdlib，TaskQueueWin，TaskQueueLibevent，TaskQueueGcd，分别对应的不同平台的具体实现

• TaskQueueStdlib 跨平台通用版本(linux下的实现，也可以用在windows环境下)
• TaskQueueWin 是windows下的实现
• TaskQueueLibevent 基于libevent实现
• TaskQueueGcd 是IOS下的实现

平台相关的实现主要是在线程的实现(webrtc中并没有使用C++ 11中thread，是自己封装了thread的API)和定时功能的实现，TaskQueueLibevent通过libevent实现定时，TaskQueueWin通过windows的消息队列实现定时。TaskQueueStdlib则是通过条件变量超时机制实现的定时。

TaskQueueBase对象创建

TaskQueueBase通过工厂类TaskQueueFactory来创建具体类型的实例，这个工厂类它还有一个工厂方法，用来创建不同的工厂类实例(有点绕- -！)

task_queue_factory.h中定义了创建TaskQueue的工厂类TaskQueueFactory，有多个工厂方法创建工厂类TaskQueueFactory。

如下代码，通过工厂类task_queue_factory创建一个TaskQueueBase对象，用于构造TaskQueue对象encoder_queue_。

//创建一个TaskQueue对象
rtc::TaskQueue encoder_queue_(task_queue_factory->CreateTaskQueue(
    "EncoderQueue",
    TaskQueueFactory::Priority::NORMAL));
1
2
3
4

用途

一个TaskQueue对象产生一个Task执行线程，它本质上是一个异步任务队列，将任务从一个线程分发至任务执行队列，可以起到线程隔离的作用，而减少同步的需求，比如将一系列关联方法放入任务队列可以减少互斥的使用，通过isCurrent判断方法是否运行在任务执行线程 ，如果是则无需添加同步措施。

可以将TaskQueue作为特定用途的线程，比如用作编码线程，将编码任务放入线程中。在webrtc中的视频编码就是如此。

代码实例

TaskQueue的使用比较简单，先通过工厂类创建一个TaskQueueBase的实例，再根据需求调用PostTask或PostDelayTask。在VideoStreamEncoder类中有个encoder_queue_的TaskQueue对象，示例如下:

//创建一个TaskQueue对象
rtc::TaskQueue encoder_queue_(task_queue_factory->CreateTaskQueue(
    "EncoderQueue",
    TaskQueueFactory::Priority::NORMAL));


//PostTask传入了一个lambada表达式，执行SendKeyFrame方法
encoder_queue_.PostTask([this] { SendKeyFrame(); });
1
2
3
4
5
6
7
8