首页 > 代码库 > 【转】linux下C语言使用线程池（附带编码）

【转】linux下C语言使用线程池（附带编码）

2024-08-08 05:37:13   224人阅读

原文链接：

linux下C语言使用线程池（附带编码） - china_sky - 博客频道 - CSDN.NET
http://blog.csdn.net/feige2008/article/details/7827390

 

 

使用一个东西，我们要明白为什么使用它，如何使用它，使用它能达到什么效果

在写本文章时，我也借鉴了网上的部分资源，因为是之前很早搜索到的资料无法追踪源头，所以在此不再写来源，谨感谢各位大神。

1    使用线程池的原因

通常使用多线程都是在需要的时候创建一个新的线程，然后执行任务，完成后退出。一般情况下是完全够满足我们的程序的。

但是当我们需要创建大量的线程，并且执行一个简单的任务之后销毁，比如：在web，email，db里面的一些应用，如彩铃，或者网络通信编程，或者云计算里面后台镜像处理的时候，我们的应用在任何时候都要准备面对数目巨大的连接请求，同时，这些请求执行的任务却又比较简单，占用的时间很少，这样我们可能就会处于不停的创建线程并销毁线程的状态。虽说比起进程的创建，线程的创建时间已经大大缩短，但是如果需要频繁的创建线程，并且每个线程所占用的处理时间又非常简短，则线程创建和销毁带给处理器的额外负担也是很可观的。

线程池的作用正是在这种情况下有效的降低频繁创建销毁线程所带来的额外开销。一般来说，线程池都是采用预创建的技术，在应用启动之初便预先创建一定数目的线程。应用在运行的过程中，需要时可以从这些线程所组成的线程池里申请分配一个空闲的线程，来执行一定的任务，任务完成后，并不是将线程销毁，而是将它返还给线程池，由线程池自行管理。如果线程池中预先分配的线程已经全部分配完毕，但此时又有新的任务请求，则线程池会动态的创建新的线程去适应这个请求。当然，有可能，某些时段应用并不需要执行很多的任务，导致了线程池中的线程大多处于空闲的状态，为了节省系统资源，线程池就需要动态的销毁其中的一部分空闲线程。因此，线程池都需要一个管理者，按照一定的要求去动态的维护其中线程的数目。

 当然，如果线程创建和销毁时间相比任务执行时间可以忽略不计，则没有必要使用线程池了。

 

2 如何使用它

线程池会维护一个任务链表(每个CThread_worker结构就是一个任务)。
pool_init()函数预先创建好max_thread_num个线程，每个线程执thread_routine ()函数。该函数中

  1. while (pool->cur_queue_size == 0)
  2. {
  3.        pthread_cond_wait (&(pool->queue_ready),&(pool->queue_lock));
  4. }

表示如果任务链表中没有任务，则该线程出于阻塞等待状态。否则从队列中取出任务并执行。
   
   pool_add_worker()函数向线程池的任务链表中加入一个任务，加入后通过调用pthread_cond_signal (&(pool->queue_ready))唤醒一个出于阻塞状态的线程(如果有的话)。
   
   pool_destroy ()函数用于销毁线程池，线程池任务链表中的任务不会再被执行，但是正在运行的线程会一直把任务运行完后再退出

 

 

3 使用它能达到的效果

 

实际上，创建太多的线程可能会导致由于过度使用系统资源而耗尽内存。为了防止资源不足，服务器应用程序可以采用线程池来限制同一时刻处理的线程数目。线程池不会占用主线程，也不会延迟后续请求的处理。一旦池中的某个线程完成任务，它将返回到等待线程队列中，等待被再次使用。这种重用使应用程序可以避免为每个任务创建新线程引起的资源和时间消耗。
4  代码实现
这是个比较简单的版本，如果想更好的话，需要再改进一年，推荐大家去网上找一下tinyftp的源码，里面有个thread pool的实现很完善很好。
[cpp] view plaincopyprint?
 
  1. <pre name="code" class="cpp">/*********************************************/  
  2. //  
  3. //          DATE:   2012-8-3  
  4. //          TIME:   下午2:42:21  
  5. //          author: sky  
  6. //          file:   simple_pool.c  
  7. //          todo:   TODO  
  8. //  
  9. /**********************************************/  
  10. #include <stdlib.h>  
  11. #include <stdio.h>  
  12. #include <unistd.h>  
  13. #include <pthread.h>  
  14. #include <assert.h>  
  15.   
  16. /* 
  17. *线程池里所有运行和等待的任务都是一个Thread_worker 
  18. *由于所有任务都在链表里，所以是一个链表结构 
  19. */  
  20. typedef struct worker  
  21. {  
  22.      /*回调函数，任务运行时会调用此函数，注意也可声明成其它形式*/  
  23.     void *(*process_interface)(void * arg);  
  24.     void *arg;  
  25.     struct worker * next;  
  26.   
  27. } Thread_worker;  
  28.   
  29. typedef struct  
  30. {  
  31.     pthread_mutex_t pool_mutex;  
  32.     pthread_cond_t pool_cond;  
  33.   
  34.     /*任务队列的头指针*/  
  35.     Thread_worker *worker_queue_head;  
  36.   
  37.     /*是否销毁线程池*/  
  38.     int is_destory;  
  39.     pthread_t * thread_id;  
  40.     int thread_max_num;  
  41.     int cur_queue_size;  
  42.   
  43. } Thread_pool;  
  44.   
  45. //此处声明了一个pool的全局变量  
  46. Thread_pool *pool = NULL;  
  47.   
  48. //////////////////////////////////  
  49. //  
  50. //function:pool_init  
  51. //arguments:thread_max_num  
  52. //TODO:初始化连接池  
  53. //  
  54. ///////////////////////////////////  
  55. void pool_init(int thread_max_num);  
  56.   
  57. //////////////////////////////////  
  58. //  
  59. //function:pool_add_worker  
  60. //arguments:void *(*process)(void *args), void *args  
  61. //TODO:添加任务  
  62. //  
  63. ///////////////////////////////////  
  64. int pool_add_worker(void *(*process)(void *args), void *args);  
  65.   
  66. void *thread_routine(void * args);  
  67.   
  68. //////////////////////////////////  
  69. //  
  70. //function:pool_destory  
  71. //arguments:  
  72. //TODO:销毁连接池  
  73. //  
  74. ///////////////////////////////////  
  75. int pool_destory();  
  76.   
  77. //////////////////////  
  78. //具体实现如下  
  79. ///////////////////////  
  80.   
  81. void pool_init(int thread_max_num)  
  82. {  
  83.     pool = (Thread_pool *) malloc(sizeof(Thread_pool));  
  84.   
  85.     pthread_mutex_init(&(pool->pool_mutex), NULL );  
  86.     pthread_cond_init(&(pool->pool_cond), NULL );  
  87.   
  88.     pool->is_destory = 0; //0代表不销毁  
  89.     pool->worker_queue_head = NULL;  
  90.     pool->thread_max_num = thread_max_num;  
  91.     pool->cur_queue_size = 0;  
  92.     pool->thread_id = (pthread_t *) malloc(sizeof(pthread_t) * thread_max_num);  
  93.     int i = 0;  
  94.     for (; i < thread_max_num; i++)  
  95.     {  
  96.         pthread_create(&(pool->thread_id[i]), NULL, thread_routine, NULL );  
  97.     }  
  98.   
  99. }  
  100.   
  101. void * thread_routine(void *args)  
  102. {  
  103.     printf("starting thread 0x%x\n", pthread_self());  
  104.     while (1)  
  105.     {  
  106.         /*如果等待队列为0并且不销毁线程池，则处于阻塞状态; 注意 
  107.          pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁*/  
  108.         pthread_mutex_lock(&(pool->pool_mutex));  
  109.         while (pool->cur_queue_size == 0 && !pool->is_destory)  
  110.         {  
  111.             printf("thread 0x%x is waiting\n", pthread_self());  
  112.             pthread_cond_wait(&(pool->pool_cond), &(pool->pool_mutex));  
  113.         }  
  114.           /*线程池要销毁了*/  
  115.         if (pool->is_destory)  
  116.         {  
  117.             pthread_mutex_unlock(&(pool->pool_mutex));  
  118.             printf("thread 0x%x will exit\n", pthread_self());  
  119.             pthread_exit(NULL );  
  120.         }  
  121.         printf("thread 0x%x is starting to work\n", pthread_self());  
  122.   
  123.         assert(pool->cur_queue_size!=0);  
  124.         assert(pool->worker_queue_head!=NULL);  
  125.   
  126.         pool->cur_queue_size--;  
  127.         Thread_worker *worker = pool->worker_queue_head;  
  128.         pool->worker_queue_head = worker->next;  
  129.         pthread_mutex_unlock(&(pool->pool_mutex));  
  130.         /*调用回调函数，执行任务*/  
  131.         (*(worker->process_interface))(worker->arg);  
  132.         free(worker);  
  133.         worker = NULL;  
  134.     }  
  135.     pthread_exit(NULL );  
  136. }  
  137.   
  138. int pool_add_worker(void *(*process)(void * args), void*args)  
  139. {  
  140.     Thread_worker *new_worker = (Thread_worker *) malloc(sizeof(Thread_worker));  
  141.     new_worker->process_interface = process;  
  142.     new_worker->arg = args;  
  143.     new_worker->next = NULL;  
  144.   
  145.       /*将任务加入到等待队列中*/  
  146.     pthread_mutex_lock(&(pool->pool_mutex));  
  147.     Thread_worker *member = pool->worker_queue_head;  
  148.     if (member != NULL )  
  149.     {  
  150.         while (member->next != NULL )  
  151.             member = member->next;  
  152.         member->next = new_worker;  
  153.     }  
  154.     else  
  155.     {  
  156.         pool->worker_queue_head = new_worker;  
  157.     }  
  158.     assert(pool->worker_queue_head!=NULL);  
  159.   
  160.     pool->cur_queue_size++;  
  161.     pthread_mutex_unlock(&(pool->pool_mutex));  
  162.      /*好了，等待队列中有任务了，唤醒一个等待线程；*/  
  163.     pthread_cond_signal(&(pool->pool_cond));  
  164.     return 0;  
  165. }  
  166.   
  167. int pool_destory()  
  168. {  
  169.     if (pool->is_destory)  
  170.         return -1;  
  171.     pool->is_destory = 1;  
  172.      /*唤醒所有等待线程，线程池要销毁了*/  
  173.     pthread_cond_broadcast(&(pool->pool_cond));  
  174.   
  175.     int i;  
  176.     for (i = 0; i < pool->thread_max_num; i++)  
  177.         pthread_join(pool->thread_id[i], NULL );  
  178.     /*销毁各种变量*/  
  179.     free(pool->thread_id);  
  180.   
  181.     Thread_worker *temp = NULL;  
  182.     while (pool->worker_queue_head != NULL )  
  183.     {  
  184.         temp = pool->worker_queue_head;  
  185.         pool->worker_queue_head = pool->worker_queue_head->next;  
  186.         free(temp);  
  187.     }  
  188.   
  189.     pthread_mutex_destroy(&(pool->pool_mutex));  
  190.     pthread_cond_destroy(&(pool->pool_cond));  
  191.   
  192.     free(pool);  
  193.     pool = NULL;  
  194.     return 0;  
  195. }  
  196.   
  197. ///////////////////////////////////////  
  198. //下面的代码就是如何调用thread pool  
  199. ///////////////////////////////////////  
  200.   
  201. void * my_process(void *arg)  
  202. {  
  203.     printf("Thread_id is 0x%x , working on task %d\n", pthread_self(),  
  204.             *(int *) arg);  
  205.   
  206.     sleep(1);  
  207.     return NULL ;  
  208. }  
  209.   
  210. int main(int argc, char *argv[])  
  211. {  
  212.     pool_init(3);  
  213.   
  214.     int *working_num = (int *) malloc(sizeof(int) * 5);  
  215.     int i = 0;  
  216.     for (; i < 5; i++)  
  217.     {  
  218.         working_num[i] = i;  
  219.         pool_add_worker(my_process, &working_num[i]);  
  220.     }  
  221.   
  222.     sleep(5);  
  223.     pool_destory();  
  224.     free(working_num);  
  225.     return 0;  
  226. }  
  227. </pre><br><br>  
5 运行测试
 
$ gcc -o threadpool threadpool.c -lpthread
【转】linux下C语言使用线程池（附带编码）






























        

            
                
                  
                
                

                    联系

                    我们