UtilBox(ub)基础组件 -- 并发任务队列Taskqueue/TaskDispatcher (一)

        Task dispatcher， 任务分发队列，也可以叫做Taskqueue。之前好多同学看到这篇文章内容是“.....”，我是想做一个标记，想写这篇文章，但是没填内容，看的同学还挺多的，以为我是标题党，这里首先表示一下歉意。下次不会了哈。

        1. Taskqueue模型简介

        2. Taskqueue的用处和优势

        3. Taskqueue适用场景

        4. 实现的技术点

        1. 模型简介

言归正传，任务队列，顾名思义，就是一个可以放任务并执行的队列。试想这么一个场景，技术经理给他下面的开发人员一些开发任务，这些任务之间没有联系，都是相互独立的小模块，经理把这些小任务按顺序放到写在卡片上并放粘贴在小黑板上，这些开发人员一个一个的从黑板上领取自己的任务，并摘下卡片，然后去闷头开发，最后做完了把结果放回去。在这个场景里，经理只管往小黑板上放任务就可以了，没任务可发放了就去喝咖啡歇着。开发人员只管从小黑板上领任务去开发就行了，不用管经理，没任务做了，就可以"度假"去了(虽然不会真的放假哈)。

        可以看出，经理和开发人员只要关注小黑板这个公共地方就可以了，实际上并不需要知道对方的存在，只要做好自己的任务。这就是典型的 master+worker thread 队列分发模式。就是咱们今天所说的Taskqueue(Task Dispatcher)，之后不对这两个概念做本质区分。先来看一个图：



         Taskqueue整体由三部分组成，Master线程、线程安全的Queue和多个worker进程，worker进程的个数可以根据cpu个数进行配置。比如8核cpu我们可以设置worker为6。这样可以并行的取任务然后执行。衔接两个模块的queue需要时线程安全的，才能保证任务的唯一性和worker的并发。

        归纳一下大体流程：

        1、Master线程接到任务(比如网卡IO读写任务，磁盘读取任务，耗CPU计算的任务)，把任务放到queue中。

        2、某个等待任务的worker线程会被通知接受任务并从队列里取出。

        3、在线程中执行任务。

        4、调用Task->Callback()函数，此函数是Master设置的任务完成回调函数，比如IO读取后打印出来。

        

        2. 用处和优势

        通过上图可以看出，首先，Taskqueue是个多线程的模型，多个worker可以跑在多个cpu core上，因为worker之间不存在交互和锁，这就发挥了多核优势可以并行执行。

        第二，在多线程下，可以增加任务的吞吐量。比如我们有个N个线程，如果没有queue的话，现在有N个任务到来，没有worker可以做了，那么结果就是要么拒绝执行，要么等待某个线程空闲出来。Taskqueue通过queue来充当buffer的作用，将大量的任务缓冲在这里等待执行。master的作用就是接受任务并存下来，所以系统的吞吐量取决于master的接收速度(注意是只接受)和queue的最大长度。当量任务可以被接下来(哪怕之后慢慢做)。

          第三，可以实现“异步化”。master线程从前面的逻辑接收到任务后，放入queue中就可以返回给调用者了，并不会阻塞在任务的执行阶段。而是只告诉调用者，任务接下来了，什么时候做完了就会回调之前设置的Callback。所以不会阻塞。实现了逻辑上的异步化，让调用者可以做其他逻辑。

        3. 使用场景

        估计大家对他的适用的地方也有了了解，它适用于这种任务消耗时间稍大，并且不需要同步的取到结果的逻辑。只需要扔到给master，然后就可以干其他事情了。举个例子，我们下载批量下载一些种子文件，打开迅雷，种子会被分解成一个个下载小文件的任务，然后丢给迅雷去做下载，这时候你就可以继续上网干别的了。

        再举一个例子，比如我们的c++代码调用新浪微博的接口来发送微博。我们有个publishWeibo(std::string content) ，每个content是一条微博。现在我们要发送10条微博，我们可能要for循环10次然后调用函数。这时候进程会阻塞在这里，等网络的应答。我们可以用任务队列的方式，直接把publishWeibo封装成一个task，然后扔给Master线程，Master线程就会立即返回不用等待发微博网络的应答，丢给后台的worker线程慢慢做。

        4. 实现的技术点

        1. 首先我们的master线程通过pthread_create要初始化一些worker线程，然后线程执行的回调函数就是一个死循环执行任务。

for(int i=0;i!=worker_thread_num;i++) {
pthread_t pid;
pthread_create(&pid,NULL,loop,this);
}

void* loop(void* context) {
....

while(!stop) {
task = queue.pop();
task->execute();
task->callback();
}

.....
}

        2. master要提供一个post函数接口，用来接收发过来的任务，并存入queue中，并通知worker来取

void Taskqueue::post(task) {
queue.push(task);
// 通过linux的“条件变量”方式，通知拥有这个条件变量的某个线程，
pthread_cond_signal(this->notifier);

}

        Baidu云的bae和新浪的sae都提供了这样类似的接口，也可以看一下他们的文档适用一下。接下来的文章会讲一下任务队列的具体实现。