非常精简的Linux线程池实现

非常精简的线程池实现（一）——使用互斥锁和条件变量线程池的含义跟它的名字一样，就是一个由许多线程组成的池子。有了线程池，在程序中使用多线程变得简单。我们不用再自己去操心线程的创建、撤销、管理问题，有什么要消耗大量时间的任务通通直接扔到线程池里就好了，然后我们的主程序（主线程）可以继续干自己的事去，线程池里面的线程会自动去执行这些任务。另一方面，线程池提升了多线程程序的性能。我们不需要在大量任务需要执行时现创建大量线程，然后在任务结束时又销毁大量线程，因为线程池里面的线程都是现成的而且能够重复使用。一个理想的线程池能够合理地动态调节池内线程数量，既不会因为线程过少而导致大量任务堆积，也不会因为线程过多了而增加额外的系统开销。线程池看上去很神奇的样子，那它是怎么实现的呢？线程这么虚渺在的东西也能像有形的物品一样圈在一个池子里？在只知道线程池这个名字的时候，我心里的疑惑就是这样的。其实线程池的原理非常简单，它就是一个非常典型的生产者消费者同步问题。如果不知道我说的这个问题也不要紧，我下面就解释。根据刚才描述的线程池的功能，可以看出线程池至少有两个主要动作，一个是主程序不定时地向线程池添加任务，另一个是线程池里的线程领取任务去执行。且不论任务和执行任务是个什么概念，但是一个任务肯定只能分配给一个线程执行。这样就可以简单猜想线程池的一种可能的架构了：主程序执行入队操作，把任务添加到一个队列里面；池子里的多个工作线程共同对这个队列试图执行出队操作，这里要保证同一时刻只有一个线程出队成功，抢夺到这个任务，其他线程继续共同试图出队抢夺下一个任务。所以在实现线程池之前，我们需要一个队列，我为这个线程池配备的队列单独放到了另一篇博客一个通用纯队列的实现中。这里的生产者就是主程序，生产任务（增加任务），消费者就是工作线程，消费任务（执行、减少任

务）。因为这里涉及到多个线程同时访问一个队列的问题，所以我们需要互斥锁来保护队列，同时还需要条件变量来处理主线程通知任务到达、工作线程抢夺任务的问题。如果不熟悉条件变量，我在另一篇博客LinuxC语言多线程库Pthread中条件变量的的正确用法逐步详解中作了详细说明。准备工作都差不多了，可以开始设计线程池了。一个最简单线程池应该有什么功能呢？对于使用者来说，除了创建和销毁线程池，最简单的情况下只需要一个功能一一添加任务。对于线程池自己来说，最简单的情况下不需要动态调节线程数量，不需要考虑线程同步、线程死锁等等一大堆麻烦的问题。所以最后的线程池定义为：创建线程池时指定线程池中应该固定包含多少工作线程，添加任务就是向线程池添加一个任务函数指针和任务函数需要的参数——这跟线程库中的普通线程创建函数是一样针和任务函数需要的参数——这跟线程库中的普通线程创建函数是一样的。根据这套线程池 ，我们使用线程池的应用程序应该是这个套路：#include<unistd.h>#include<pthread.h>void*test(void*arg){inti;for(i=0;i<5;i++){printf("tid:%ldtask:%ld\n",pthread_self(),(long)arg);fflush(stdout);sleep(2);}returnNULL;}intmain(){longi=0;thread_pool_tpool;pool=thread_pool_create(2);for(i=0;i<5;i++){thread_pool_add_task(pool,test,(void*)i);}puts("pressentertoterminate...");getchar();thread_pool_destroy(pool);return0;

上面这个测试程序向线程池添加了个相同的任务，每个任务耗时秒，但是线程池中只有个工作线呈，所以程序的运行结果是两个工作线程轮流把个任务挨个做完。显示到屏幕上就是：前秒两个工作线程轮流输出自己的线程和当前任务的任务号和，各输出次；第二个秒两个工作线程轮流输出自己的线程和当前任务的任务号和3……在这期间，主程序输出“pressentertoterminate...”并等待用户输入，任何时候都可以按回车让主程序继续往下，这样会强制终止所有工作线程并销毁线程池，最后程序退出。test程序运行效果截图如下：最后就是线程池真正的实现了：treapoo.inetreapooin eee.inesti.in eptrea.str ttreapoonsineinttrea ontptreattreaseettassptreamtetoptreaonttasreaptreaonttasreastructtask{void*(*routine)(void*arg);void*arg;}；staticvoidcleanup(pthread_mutex_t*lock){pthread_mutex_unlock(lock);}staticvoid*worker(thread_pool_tpool){structtask*t;while(1){pthread_mutex_lock(&pool->lock);pthread_cleanup_push((void(*)(void*))cleanup,&pool->lock);while(queue_isempty(pool->tasks)){pthread_cond_wait(&pool->task_ready,&pool->lock);/*Aconditionwait(whethertimedornot)isacancellationpoint...aside-effectofactinguponacancellationrequestwhileinaconditionwaitisthatthemutexis(ineffect)re-acquiredbeforecallingthefirstcancellationcleanuphandler.*/}t=(structtask*)queue_dequeue(pool->tasks);pthread_cleanup_pop(0);pthread_mutex_unlock(&pool->lock);t->routine(t->arg);/*todo:reportreturnedvalue*/free(t);}returnNULL;}thread_pool_tthread_pool_create(unsignedintthread_count){unsignedinti;thread_pool_tpool=NULL;pool=(thread_pool_t)malloc(sizeof(structthread_pool));pool->thread_count=thread_count;pool->threads=(pthread_t*)malloc(sizeof(pthread_t)*thread_count);pool->tasks=queue_create();pthread_mutex_init(&pool->lock,NULL);pthread_cond_init(&pool->task_ready,NULL);for(i=0;i<thread_count;i++){pthread_create(pool->threads+i,NULL,(void*(*)(void*))worker,pool);}returnpool;}voidthread_pool_add_task(thread_pool_tpool,void*(*routine)(void*arg),void*arg){structtask*t;pthread_mutex_lock(&pool->lock);t=(structtask*)queue_enqueue(pool->tasks,sizeof(structtask));上面的 函数就是工作线程函数，所有的工作线程都在执行着这个函数。它首先在互斥锁和条件变量的保护下从任务队列中取出一个任务，这个任务实际上是一个函数指针和调用函数所需的参数，所以执行任务就很简单了——用任务参数调用任务函数。函数返回以后，工作线程继续去抢任务。这里没有处理任务函数的返回值问题，理论上任务函数返回以后线程池应该用某种机制通知主程序，然后主程序获取通过某种手段获取返回值，但这明显不是一个最简单的线程池需要操心的事。实际上，应用程序可以通过全局变量或传入的参数指针，加上额外的线程同步代码解决返回值的通知和获取问题。还有一点需要注意，最后线程池销毁