线程池是怎么一回事

http://blog.csdn.net/denny_233/article/details/40484365

2014

所谓线程池，就是程序的初始化阶段，就预先创建一批线程，每个线程都做好准备干活；

然后初始化一个任务列表，当有任务来了，就往任务列表里面添加；

任务列表里面有任务了，这时候那些等待的线程们就要抢活干了，怎么抢，使用各种线程同步手段（互斥量，临界区等），人品好的线程抢到任务后，从任务列表取出任务，就可以开始干活了。干完以后，就又继续回到初始等待状态，准备抢夺下一个任务。

/**********************************************************************************************************************************************/
这样就好比你有一批小弟排队在那里等着，一旦有任务，他们会很守纪律的去抢着干，每个任务都会被一个小弟抢走，干完以后，小弟不用休息，继续等着抢下一个任务干活。这样当你的任务源源不断的到达，你的小弟们就一个个争先恐后的抢过来完成，绝不偷懒。

相 反，如果不使用线程池，每次等到任务来了，再临时创建线程。这样就相当于每次有任务时，你再临时招聘一个小弟过来，小弟完成任务后，就回家了。然后下次再有任务，又招聘一个小弟过来，完成任务后，回家。相比线程池，中间招聘小弟的时间就要额外耗费时间和精力了（创建和销毁线程中，cpu的时间，内存的分配）。/*********************************************************************************************************************************************/

总结来说，线程池有4个组成部分

线程池管理器（ThreadPoolManager）:用于创建并管理线程池

工作线程（WorkThread）: 线程池中线程

任务接口（Task）:每个任务必须实现的接口，以供工作线程调度任务的执行。

任务队列:用于存放没有处理的任务。提供一种缓冲机制。

 

所以，使用线程池，就省去了哪些额外的线程开销，从而连续的完成所有的任务。当然，线程池中用于线程同步的操作同样也有一定的消耗，但这个消耗是相对小的。另外，还可以对线程池中的线程根据当前的任务量进行动态的调整，从而更好的节省相关资源。

什么时候适合用线程池：

1. 需要大量的线程来完成，且完成时间比较段。比如WEB服务器完成网页的请求，使用线程池技术非常适合。对于长时间的任务，如telnet链接什么的，使用线程池就没有什么有点了。

2、对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。

3、接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。短时间内产生大量线程可能使内存达到极致，并出现”outofMemory”情况。

 

下面是一个网上的代码用例:

thpool.h代码如下：

[cpp] view plain copy

1. /**********************************  
2.  * @author      Johan Hanssen Seferidis 
3.  * @date        12/08/2011 
4.  * Last update: 01/11/2011 
5.  * License:     LGPL 
6.  *  
7.  **********************************/  
8.   
9. /* Description: Library providing a threading pool where you can add work on the fly. The number 
10.  *              of threads in the pool is adjustable when creating the pool. In most cases 
11.  *              this should equal the number of threads supported by your cpu. 
12.  *           
13.  *              For an example on how to use the threadpool, check the main.c file or just read 
14.  *              the documentation. 
15.  *  
16.  *              In this header file a detailed overview of the functions and the threadpool logical 
17.  *              scheme is present in case tweaking of the pool is needed.  
18.  * */  
19.   
20. /*  
21.  * Fast reminders: 
22.  *  
23.  * tp           = threadpool  
24.  * thpool       = threadpool 
25.  * thpool_t     = threadpool type 
26.  * tp_p         = threadpool pointer 
27.  * sem          = semaphore 
28.  * xN           = x can be any string. N stands for amount 
29.  *  
30.  * */  
31.                     
32. /*              _______________________________________________________         
33.  *             /                                                       \ 
34.  *             |   JOB QUEUE        | job1 | job2 | job3 | job4 | ..   | 
35.  *             |                                                       | 
36.  *             |   threadpool      | thread1 | thread2 | ..           | 
37.  *             \_______________________________________________________/ 
38.  *  
39.  *    Description:       Jobs are added to the job queue. Once a thread in the pool 
40.  *                       is idle, it is assigned with the first job from the queue(and 
41.  *                       erased from the queue). It's each thread's job to read from  
42.  *                       the queue serially(using lock) and executing each job 
43.  *                       until the queue is empty. 
44.  *    描述:              每来一个Job，会加入的队列中去，当池子中有空闲线程时，从队列中取出job 
45.  *  
46.  *    Scheme: 
47.  *  
48.  *    thpool______                jobqueue____                      ______  
49.  *    |           |               |           |       .----------->|_job0_| Newly added job 
50.  *    |           |               |  head------------'             |_job1_| 
51.  *    | jobqueue----------------->|           |                    |_job2_| 
52.  *    |           |               |  tail------------.             |__..__|  
53.  *    |___________|               |___________|       '----------->|_jobn_| Job for thread to take 
54.  *  
55.  *  
56.  *    job0________  
57.  *    |           | 
58.  *    | function---->          //Job具有通用的接口，供工作线程调度试用 
59.  *    |           | 
60.  *    |   arg-------> 
61.  *    |           |         job1________  
62.  *    |  next-------------->|           | 
63.  *    |___________|         |           |.. 
64.  */  
65.   
66. #ifndef _THPOOL_  
67. #define _THPOOL_  
68.   
69. #include <pthread.h>  
70. #include <semaphore.h>  
71.   
72. /* ================================= STRUCTURES ================================================ */  
73. /* Individual job */  
74. typedef struct thpool_job_t{  
75.     void*  (*function)(void* arg);                     /**< function pointer         */  
76.     void*                     arg;                     /**< function's argument      */  
77.     struct thpool_job_t*     next;                     /**< pointer to next job      */  
78.     struct thpool_job_t*     prev;                     /**< pointer to previous job  */  
79. }thpool_job_t;  
80.   
81. /* Job queue as doubly linked list */  
82. typedef struct thpool_jobqueue{  
83.     thpool_job_t *head;                                /**< pointer to head of queue */  
84.     thpool_job_t *tail;                                /**< pointer to tail of queue */  
85.     int           jobsN;                               /**< amount of jobs in queue  */  
86.     sem_t        *queueSem;                            /**< semaphore(this is probably just holding the same as jobsN) */  
87. }thpool_jobqueue;  
88.   
89. /* The threadpool */  
90. typedef struct thpool_t{  
91.     pthread_t*       threads;                          /**< pointer to threads' ID   */  
92.     int              threadsN;                         /**< amount of threads        */  
93.     thpool_jobqueue* jobqueue;                         /**< pointer to the job queue */  
94. }thpool_t;  
95.   
96. /* Container for all things that each thread is going to need */  
97. typedef struct thread_data{                              
98.     pthread_mutex_t *mutex_p;  
99.     thpool_t        *tp_p;  
100. }thread_data;  
101.   
102.   
103. /* =========================== FUNCTIONS ================================================ */  
104. /* ----------------------- Threadpool specific --------------------------- */  
105.   
106. /** 
107.  * @brief  Initialize threadpool 
108.  *  
109.  * Allocates memory for the threadpool, jobqueue, semaphore and fixes  
110.  * pointers in jobqueue. 
111.  *  
112.  * @param  number of threads to be used 
113.  * @return threadpool struct on success, 
114.  *         NULL on error 
115.  */  
116. thpool_t* thpool_init(int threadsN);  
117.   
118. /** 
119.  * @brief What each thread is doing 
120.  *  
121.  * In principle this is an endless loop. The only time this loop gets interuppted is once 
122.  * thpool_destroy() is invoked. 
123.  *  
124.  * @param threadpool to use 
125.  * @return nothing 
126.  */  
127. void thpool_thread_do(thpool_t* tp_p);  
128.   
129. /** 
130.  * @brief Add work to the job queue 
131.  *  
132.  * Takes an action and its argument and adds it to the threadpool's job queue. 
133.  * If you want to add to work a function with more than one arguments then 
134.  * a way to implement this is by passing a pointer to a structure. 
135.  *  
136.  * ATTENTION: You have to cast both the function and argument to not get warnings. 
137.  *  
138.  * @param  threadpool to where the work will be added to 
139.  * @param  function to add as work 
140.  * @param  argument to the above function 
141.  * @return int 
142.  */  
143. int thpool_add_work(thpool_t* tp_p, void *(*function_p)(void*), void* arg_p);  
144.   
145. /** 
146.  * @brief Destroy the threadpool 
147.  *  
148.  * This will 'kill' the threadpool and free up memory. If threads are active when this 
149.  * is called, they will finish what they are doing and then they will get destroyied. 
150.  *  
151.  * @param threadpool a pointer to the threadpool structure you want to destroy 
152.  */  
153. void thpool_destroy(thpool_t* tp_p);  
154.   
155.   
156. /* ------------------------- Queue specific ------------------------------ */  
157. /** 
158.  * @brief Initialize queue 
159.  * @param  pointer to threadpool 
160.  * @return 0 on success, 
161.  *        -1 on memory allocation error 
162.  */  
163. int thpool_jobqueue_init(thpool_t* tp_p);  
164.   
165. /** 
166.  * @brief Add job to queue 
167.  *  
168.  * A new job will be added to the queue. The new job MUST be allocated 
169.  * before passed to this function or else other functions like thpool_jobqueue_empty() 
170.  * will be broken. 
171.  *  
172.  * @param pointer to threadpool 
173.  * @param pointer to the new job(MUST BE ALLOCATED) 
174.  * @return nothing  
175.  */  
176. void thpool_jobqueue_add(thpool_t* tp_p, thpool_job_t* newjob_p);  
177.   
178. /** 
179.  * @brief Remove last job from queue.  
180.  *  
181.  * This does not free allocated memory so be sure to have peeked() \n 
182.  * before invoking this as else there will result lost memory pointers. 
183.  *  
184.  * @param  pointer to threadpool 
185.  * @return 0 on success, 
186.  *         -1 if queue is empty 
187.  */  
188. int thpool_jobqueue_removelast(thpool_t* tp_p);  
189.   
190. /**  
191.  * @brief Get last job in queue (tail) 
192.  *  
193.  * Gets the last job that is inside the queue. This will work even if the queue 
194.  * is empty. 
195.  *  
196.  * @param  pointer to threadpool structure 
197.  * @return job a pointer to the last job in queue, 
198.  *         a pointer to NULL if the queue is empty 
199.  */  
200. thpool_job_t* thpool_jobqueue_peek(thpool_t* tp_p);  
201.   
202. /** 
203.  * @brief Remove and deallocate all jobs in queue 
204.  *  
205.  * This function will deallocate all jobs in the queue and set the 
206.  * jobqueue to its initialization values, thus tail and head pointing 
207.  * to NULL and amount of jobs equal to 0. 
208.  *  
209.  * @param pointer to threadpool structure 
210.  * */  
211. void thpool_jobqueue_empty(thpool_t* tp_p);  
212.   
213. #endif  
214. </span>  

thpool.c代码如下

[cpp] view plain copy

1. /* ******************************** 
2.  *  
3.  * Author:  Johan Hanssen Seferidis 
4.  * Date:    12/08/2011 
5.  * Update:  01/11/2011 
6.  * License: LGPL 
7.  *  
8.  *  
9.  *//** @file thpool.h *//* 
10.  ********************************/  
11.   
12. /* Library providing a threading pool where you can add work. For an example on  
13.  * usage you refer to the main file found in the same package */  
14.   
15. /*  
16.  * Fast reminders: 
17.  *  
18.  * tp           = threadpool  
19.  * thpool       = threadpool 
20.  * thpool_t     = threadpool type 
21.  * tp_p         = threadpool pointer 
22.  * sem          = semaphore 
23.  * xN           = x can be any string. N stands for amount 
24.  *  
25.  * */  
26.   
27. #include <stdio.h>  
28. #include <stdlib.h>  
29. #include <pthread.h>  
30. #include <semaphore.h>  
31. #include <errno.h>  
32.   
33. #include "thpool.h"      /* here you can also find the interface to each function */  
34.   
35.   
36. static int thpool_keepalive=1;  
37.   
38. /* Create mutex variable */  
39. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; /* used to serialize queue access */  
40.   
41.   
42. /* Initialise thread pool */  
43. thpool_t* thpool_init(int threadsN){  
44.     thpool_t* tp_p;  
45.       
46.     if (!threadsN || threadsN<1) threadsN=1;  
47.       
48.     /* Make new thread pool */  
49.     tp_p=(thpool_t*)malloc(sizeof(thpool_t));                              /* MALLOC thread pool */  
50.     if (tp_p==NULL){  
51.         fprintf(stderr, "thpool_init(): Could not allocate memory for thread pool\n");  
52.         return NULL;  
53.     }  
54.     tp_p->threads=(pthread_t*)malloc(threadsN*sizeof(pthread_t));          /* MALLOC thread IDs */  
55.     if (tp_p->threads==NULL){  
56.         fprintf(stderr, "thpool_init(): Could not allocate memory for thread IDs\n");  
57.         return NULL;  
58.     }  
59.     tp_p->threadsN=threadsN;  
60.       
61.     /* Initialise the job queue */  
62.     if (thpool_jobqueue_init(tp_p)==-1){  
63.         fprintf(stderr, "thpool_init(): Could not allocate memory for job queue\n");  
64.         return NULL;  
65.     }  
66.       
67.     /* Initialise semaphore，关于信号量的操作请参考其他*/  
68.     tp_p->jobqueue->queueSem=(sem_t*)malloc(sizeof(sem_t));                 /* MALLOC job queue semaphore */  
69.     sem_init(tp_p->jobqueue->queueSem, 0, 0); /* no shared, initial value */  
70.       
71.     /* Make threads in pool */  
72.     int t;  
73.     for (t=0; t<threadsN; t++){  
74.         printf("Created thread %d in pool \n", t);  
75.         pthread_create(&(tp_p->threads[t]), NULL, (void *)thpool_thread_do, (void *)tp_p); /* MALLOCS INSIDE PTHREAD HERE */  
76.     }  
77.       
78.     return tp_p;  
79. }  
80.   
81.   
82. /* What each individual thread is doing ，线程池的入口函数，从队列中抽取共同接口的Task;在这里即thpool_job_t的 
83. 前2个参数void*  (*function)(void* arg) 及void *arg. */  
84. /* There are two scenarios here. One is everything works as it should and second if 
85.  * the thpool is to be killed. In that manner we try to BYPASS sem_wait and end each thread. */  
86. void thpool_thread_do(thpool_t* tp_p){  
87.   
88.     while(thpool_keepalive){  
89.         //收到sem_post signal  
90.         if (sem_wait(tp_p->jobqueue->queueSem)) {/* WAITING until there is work in the queue */  
91.             perror("thpool_thread_do(): Waiting for semaphore");  
92.             exit(1);  
93.         }  
94.   
95.         if (thpool_keepalive){  
96.               
97.             /* Read job from queue and execute it */  
98.             void*(*func_buff)(void* arg);  
99.             void*  arg_buff;  
100.             thpool_job_t* job_p;  
101.       
102.             pthread_mutex_lock(&mutex);                  /* LOCK */  
103.               
104.             job_p = thpool_jobqueue_peek(tp_p);  
105.             func_buff=job_p->function;  
106.             arg_buff =job_p->arg;                     /*每个入队列的Task都具有的接口*/  
107.             thpool_jobqueue_removelast(tp_p);  
108.               
109.             pthread_mutex_unlock(&mutex);                /* UNLOCK */  
110.               
111.             func_buff(arg_buff);                         /* run function */  
112.             free(job_p);                                                       /* DEALLOC job */  
113.         }  
114.         else  
115.         {  
116.             return; /* EXIT thread*/  
117.         }  
118.     }  
119.     return;  
120. }  
121.   
122.   
123. /* Add work to the thread pool */  
124. int thpool_add_work(thpool_t* tp_p, void *(*function_p)(void*), void* arg_p){  
125.     thpool_job_t* newJob;  
126.       
127.     newJob=(thpool_job_t*)malloc(sizeof(thpool_job_t));                        /* MALLOC job */  
128.     if (newJob==NULL){  
129.         fprintf(stderr, "thpool_add_work(): Could not allocate memory for new job\n");  
130.         exit(1);  
131.     }  
132.       
133.     /* add function and argument */  
134.     newJob->function=function_p;  
135.     newJob->arg=arg_p;  
136.       
137.     /* add job to queue */  
138.     pthread_mutex_lock(&mutex);                  /* LOCK */  
139.     thpool_jobqueue_add(tp_p, newJob);  
140.     pthread_mutex_unlock(&mutex);                /* UNLOCK */  
141.       
142.     return 0;  
143. }  
144.   
145.   
146. /* Destroy the threadpool */  
147. void thpool_destroy(thpool_t* tp_p){  
148.     int t;  
149.       
150.     /* End each thread's infinite loop */  
151.     thpool_keepalive=0;   
152.   
153.     /* Awake idle threads waiting at semaphore */  
154.     for (t=0; t<(tp_p->threadsN); t++){  
155.         if (sem_post(tp_p->jobqueue->queueSem)){  
156.             fprintf(stderr, "thpool_destroy(): Could not bypass sem_wait()\n");  
157.         }  
158.     }  
159.   
160.     /* Kill semaphore */  
161.     if (sem_destroy(tp_p->jobqueue->queueSem)!=0){  
162.         fprintf(stderr, "thpool_destroy(): Could not destroy semaphore\n");  
163.     }  
164.       
165.     /* Wait for threads to finish */  
166.     for (t=0; t<(tp_p->threadsN); t++){  
167.         pthread_join(tp_p->threads[t], NULL);  
168.     }  
169.       
170.     thpool_jobqueue_empty(tp_p);  
171.       
172.     /* Dealloc */  
173.     free(tp_p->threads);                                                   /* DEALLOC threads             */  
174.     free(tp_p->jobqueue->queueSem);                                        /* DEALLOC job queue semaphore */  
175.     free(tp_p->jobqueue);                                                  /* DEALLOC job queue           */  
176.     free(tp_p);                                                            /* DEALLOC thread pool         */  
177. }  
178.   
179.   
180.   
181. /* =================== JOB QUEUE OPERATIONS ===================== */  
182.   
183. /* Initialise queue */  
184. int thpool_jobqueue_init(thpool_t* tp_p){  
185.     tp_p->jobqueue=(thpool_jobqueue*)malloc(sizeof(thpool_jobqueue));      /* MALLOC job queue */  
186.     if (tp_p->jobqueue==NULL) return -1;  
187.     tp_p->jobqueue->tail=NULL;  
188.     tp_p->jobqueue->head=NULL;  
189.     tp_p->jobqueue->jobsN=0;  
190.     return 0;  
191. }  
192.   
193. /* Add job to queue */  
194. void thpool_jobqueue_add(thpool_t* tp_p, thpool_job_t* newjob_p){ /* remember that job prev and next point to NULL */  
195.   
196.     newjob_p->next=NULL;  
197.     newjob_p->prev=NULL;  
198.       
199.     thpool_job_t *oldFirstJob;  
200.     oldFirstJob = tp_p->jobqueue->head;  
201.       
202.     /* fix jobs' pointers */  
203.     switch(tp_p->jobqueue->jobsN){  
204.           
205.         case 0:     /* if there are no jobs in queue */  
206.                     tp_p->jobqueue->tail=newjob_p;  
207.                     tp_p->jobqueue->head=newjob_p;  
208.                     break;  
209.           
210.         default:    /* if there are already jobs in queue */  
211.                     oldFirstJob->prev=newjob_p;  
212.                     newjob_p->next=oldFirstJob;  
213.                     tp_p->jobqueue->head=newjob_p;  
214.   
215.     }  
216.   
217.     (tp_p->jobqueue->jobsN)++;     /* increment amount of jobs in queue */  
218.     sem_post(tp_p->jobqueue->queueSem);  
219.       
220.     int sval;  
221.     sem_getvalue(tp_p->jobqueue->queueSem, &sval);  
222. }  
223.   
224.   
225. /* Remove job from queue */  
226. int thpool_jobqueue_removelast(thpool_t* tp_p){  
227.     thpool_job_t *oldLastJob;  
228.     oldLastJob = tp_p->jobqueue->tail;  
229.       
230.     /* fix jobs' pointers */  
231.     switch(tp_p->jobqueue->jobsN){  
232.           
233.         case 0:     /* if there are no jobs in queue */  
234.                     return -1;  
235.                     break;  
236.           
237.         case 1:     /* if there is only one job in queue */  
238.                     tp_p->jobqueue->tail=NULL;  
239.                     tp_p->jobqueue->head=NULL;  
240.                     break;  
241.                       
242.         default:    /* if there are more than one jobs in queue */  
243.                     oldLastJob->prev->next=NULL;               /* the almost last item */  
244.                     tp_p->jobqueue->tail=oldLastJob->prev;  
245.                       
246.     }  
247.       
248.     (tp_p->jobqueue->jobsN)--;  
249.       
250.     int sval;  
251.     sem_getvalue(tp_p->jobqueue->queueSem, &sval);  
252.     return 0;  
253. }  
254.   
255.   
256. /* Get first element from queue */  
257. thpool_job_t* thpool_jobqueue_peek(thpool_t* tp_p){  
258.     return tp_p->jobqueue->tail;  
259. }  
260.   
261. /* Remove and deallocate all jobs in queue */  
262. void thpool_jobqueue_empty(thpool_t* tp_p){  
263.       
264.     thpool_job_t* curjob;  
265.     curjob=tp_p->jobqueue->tail;  
266.       
267.     while(tp_p->jobqueue->jobsN){  
268.         tp_p->jobqueue->tail=curjob->prev;  
269.         free(curjob);  
270.         curjob=tp_p->jobqueue->tail;  
271.         tp_p->jobqueue->jobsN--;  
272.     }  
273.       
274.     /* Fix head and tail */  
275.     tp_p->jobqueue->tail=NULL;  
276.     tp_p->jobqueue->head=NULL;  
277. }  
278. </span>  

main.c 代码如下

[cpp] view plain copy

1. /*  
2.  * This is just an example on how to use the thpool library  
3.  *  
4.  * We create a pool of 4 threads and then add 20 tasks to the pool(10 task1  
5.  * functions and 10 task2 functions). 
6.  *  
7.  * Task1 doesn't take any arguments. Task2 takes an integer. Task2 is used to show 
8.  * how to add work to the thread pool with an argument. 
9.  *  
10.  * As soon as we add the tasks to the pool, the threads will run them. One thread 
11.  * may run x tasks in a row so if you see as output the same thread running several 
12.  * tasks, it's not an error. 
13.  *  
14.  * All jobs will not be completed and in fact maybe even none will. You can add a sleep() 
15.  * function if you want to complete all tasks in this test file to be able and see clearer 
16.  * what is going on. 
17.  *  
18.  * */  
19.   
20. #include <stdio.h>  
21.   
22. #include "thpool.h"  
23.   
24.   
25.   
26. /* Some arbitrary task 1 */  
27. void task1(){  
28.     printf("# Thread working: %u\n", (int)pthread_self());  
29.     printf("  Task 1 running..\n");  
30. }  
31.   
32.   
33.   
34. /* Some arbitrary task 2 */  
35. void task2(int a){  
36.     printf("# Thread working: %u\n", (int)pthread_self());  
37.     printf("  Task 2 running..\n");  
38.     printf("%d\n", a);  
39. }  
40.   
41.   
42.   
43. int main(){  
44.     int i;  
45.       
46.     thpool_t* threadpool;             /* make a new thread pool structure     */  
47.     threadpool=thpool_init(4);        /* initialise it to 4 number of threads */  
48.   
49.   
50.     puts("Adding 20 tasks to threadpool");  
51.     int a=54;  
52.     for (i=0; i<10; i++){  
53.         thpool_add_work(threadpool, (void*)task1, NULL);  
54.         thpool_add_work(threadpool, (void*)task2, (void*)a);  
55.     };  
56.   
57.   
58.     puts("Will kill threadpool");  
59.     thpool_destroy(threadpool);  
60.       
61.     return 0;  
62. }  
63. </span>