池化技术与线程池的简单实现

池化技术

在系统开发过程中，我们经常会用到池化技术。通俗的讲，池化技术就是：把一些资源预先分配好，组织到对象池中，之后的业务使用资源从对象池中获取，使用完后放回到对象池中。这样做带来几个明显的好处：

1），资源重复使用， 减少了资源分配和释放过程中的系统消耗。比如，在IO密集型的服务器上，并发处理过程中的子线程或子进程的创建和销毁过程，带来的系统开销将是难以接受的。所以在业务实现上，通常把一些资源预先分配好，如线程池，数据库连接池，Redis连接池，HTTP连接池等，来减少系统消耗，提升系统性能。

2），可以对资源的整体使用做限制。这个好理解，相关资源预分配且只在预分配是生成，后续不再动态添加，从而限制了整个系统对资源的使用上限。类似一个令牌桶的功能。

3），池化技术分配对象池，通常会集中分配，这样有效避免了碎片化的问题。

线程池

有了上面的池化技术铺垫，理解线程池就很容易了：

1）先启动若干数量的线程，并让这些线程都处于睡眠状态，并以一定的方式组织起来，形成“线程池”；

2）当客户端有一个新请求时，就会从线程池中选取一个线程，并唤醒线程，让它来处理客户端的这个请求；

3）当处理完这个请求后，线程又处于睡眠状态，并且放回到线程池中，等待任务到来调度。

线程池设计

 如上图，是基于线程条件变量唤醒的简单线程池模型，主要包括了：任务，任务队列，工作线程几个组件。说明如下：

任务（job）

任务是对一次可执行过的描述，通常包括执行的方法（执行函数）和数据。比如，一个报文接收过程就是一个任务，包括：报文处理函数和报文内容。本例中，对job的定义如下：

点击(此处)折叠或打开

1. typedef struct thread_job {
   
2.     struct list_head list;
   
3.     struct thread_pool* pool;
   
4.     job_process_func jfunc; // 任务执行函数
   
5.     void* data; // 任务数据，由调用者分配，框架释放
   
6. } job_t

任务队列（job_queue）

任务队列缓存着所有待执行的任务，供各个工作线程调度。当工作线程空闲或被唤醒时，会尝试从任务队列中摘取一个任务，然后在线程空间内执行。任务队列用线程互斥锁进行保护。为了方便理解整个任务执行转换的过程，本例中，把空闲任务队列单独出来了。 所以，任务的状态有四种： 
    1）在空闲对列中等待业务；
    2）在业务流程中装配；
    3）在工作队列中等待调度；

    4）在工作线程中执行。

工作线程

     工作线程等待在条件变量上（job_dequeue_valid，新任务需要处理），当新任务到达时，线程被唤醒。线程会尝试从任务队列上获取任务，并执行它；执行完毕后重新等待在条件变量上。如此往复：

点击(此处)折叠或打开

1. while (1)
   
2.     {
   
3.         if (thrdp_stopping(pool))
   
4.         {
   
5.             pthread_exit(NULL);
   
6.         }
   
7. 
   
8.         pthread_mutex_lock(&(pool->qlock));
   
9.         while (list_empty(&pool->job_queue))
   
10.         {
    
11.             // nothing to process, wait until job come in
    
12.             printf("thread get none job, hangup...\n", (unsigned int)pthread_self() );
    
13.             pthread_cond_wait(&(pool->job_dequeue_valid), &(pool->qlock));
    
14. 
    
15.             if (thrdp_stopping(pool))
    
16.             {
    
17.                 pthread_exit(NULL);
    
18.             }
    
19. 
    
20.         }
    
21.         job = list_first_entry(&pool->job_queue, struct thread_job, list);
    
22.         list_del(&job->list);
    
23.         pthread_mutex_unlock(&(pool->qlock));
    
24.         job->jfunc(job);
    
25.     }

如上代码片段，特别提醒一下，当工作线程从等待条件中被唤醒时，一定要对当前的执行条件再检查，原因是当前线程可能是被“虚假唤醒”。

比如，当前队列是空的，有A，B，C三个工作线程都在等待任务；当一个任务R0到来时，A,B,C三个线程均可能被唤醒；由于任务队列锁的关系，其中只有一个（假如A）取得的实际任务，而B和C均未取得任务，因此B和C称为被“虚假唤醒”。此时，B和C必须对任务获取状态进行再次判断，确保正常获取到任务，方能往下执行调度，否则，应该重新休眠等待任务。

代码实现

附完整代码和测试代码threadpool.h

点击(此处)折叠或打开

1. #ifndef __THREAD_POOL_H__
   
2. #define __THREAD_POOL_H__
   
3. #include <linux/types.h>
   
4. #include <pthread.h>
   
5. #include <syslog.h>
   
6. #include <sys/stat.h>
   
7. #include <unistd.h>
   
8. #include <stdio.h>
   
9. #include <stdlib.h>
   
10. #include <sys/socket.h>
    
11. #include <sys/types.h>
    
12. #include <string.h>
    
13. #include <asm/types.h>
    
14. #include <linux/netlink.h>
    
15. #include <linux/socket.h>
    
16. #include <linux/types.h>
    
17. #include <sys/socket.h>
    
18. #include <netinet/in.h>
    
19. #include <arpa/inet.h>
    
20. #include <errno.h>
    
21. #include <unistd.h>
    
22. 
    
23. #include "list.h"
    
24. 
    
25. 
    
26. #define THRDP_STATUS_STOPPING 0x01 // 线程池是否已经关闭
    
27. 
    
28. typedef void (*job_process_func)(void*);
    
29. 
    
30. struct thread_pool
    
31. {
    
32.     int thread_num; // 线程池中开启线程的个数
    
33.     int max_job_num; // 队列中最大job的个数
    
34.     struct list_head job_queue; // 待处理的job队列
    
35.     struct list_head free_job_queue; // job空闲池
    
36. 
    
37. 
    
38.     pthread_t *pthreads; //线程池中所有线程的pthread_t
    
39.     pthread_mutex_t qlock; //互斥信号量,保护job_queue
    
40.     pthread_mutex_t fqlock; //互斥信号量,保护free_job_queue
    
41.     pthread_cond_t job_dequeue_valid; // 隊列不為空,需要喚醒線程處理
    
42.     pthread_cond_t job_enqueue_valid; // 隊列沒有滿,可以繼續添加job
    
43. 
    
44.     int flags;
    
45. };
    
46. 
    
47. typedef struct thread_job {
    
48.     struct list_head list;
    
49.     struct thread_pool* pool;
    
50.     job_process_func jfunc; // 任务执行函数
    
51.     void* data; // 任务数据，由调用者分配，框架释放
    
52. } job_t;
    
53. 
    
54. struct thread_pool* THRDP_new(unsigned int thread_num, unsigned int max_job_num);
    
55. void THRDP_destroy(struct thread_pool* pool);
    
56. struct thread_job* THRDP_job_get(struct thread_pool* pool);
    
57. void THRDP_job_put(struct thread_pool* pool, struct thread_job* job);
    
58. void THRDP_job_add(struct thread_pool* pool, struct thread_job* job);
    
59. 
    
60. #endif

    注意，上面引用了list.h，请查看我之前发的用户态list的头文件。
    threadpool.c

点击(此处)折叠或打开

1. #include "threadpool.h"
   
2. 
   
3. 
   
4. static inline int thrdp_stopping (struct thread_pool* pool)
   
5. {
   
6.     return pool->flags & THRDP_STATUS_STOPPING;
   
7. }
   
8. 
   
9. static void* thrdp_process_func(void* arg)
   
10. {
    
11.     struct thread_pool* pool = (struct thread_pool*)arg;
    
12.     struct thread_job* job;
    
13. 
    
14.     while (1)
    
15.     {
    
16.         if (thrdp_stopping(pool))
    
17.         {
    
18.             pthread_exit(NULL);
    
19.         }
    
20. 
    
21.         pthread_mutex_lock(&(pool->qlock));
    
22.         while (list_empty(&pool->job_queue))
    
23.         {
    
24.             // nothing to process, wait until job come in
    
25.             printf("thread get none job, hangup...\n", (unsigned int)pthread_self() );
    
26.             pthread_cond_wait(&(pool->job_dequeue_valid), &(pool->qlock));
    
27. 
    
28.             if (thrdp_stopping(pool))
    
29.             {
    
30.                 pthread_exit(NULL);
    
31.             }
    
32. 
    
33.         }
    
34.         job = list_first_entry(&pool->job_queue, struct thread_job, list);
    
35.         list_del(&job->list);
    
36.         pthread_mutex_unlock(&(pool->qlock));
    
37.         job->jfunc(job);
    
38.     }
    
39. }
    
40. 
    
41. 
    
42. struct thread_pool* THRDP_new(unsigned int thread_num, unsigned int max_job_num)
    
43. {
    
44.     struct thread_pool* pool = NULL;
    
45.     struct thread_job* job;
    
46.     struct thread_job* tmp;
    
47.     int i;
    
48. 
    
49.     if(thread_num < 2)
    
50.     {
    
51.         thread_num = 2;
    
52.     }
    
53. 
    
54.     if (max_job_num < thread_num)
    
55.     {
    
56.         max_job_num = thread_num;
    
57.     }
    
58. 
    
59.     pool = malloc(sizeof(struct thread_pool));
    
60.     if (NULL == pool)
    
61.     {
    
62.         printf("failed to malloc thread_pool!\n");
    
63.         return NULL;
    
64.     }
    
65.     memset(pool, 0, sizeof(struct thread_pool) );
    
66.     pool->thread_num = thread_num;
    
67.     pool->max_job_num = max_job_num;
    
68. 
    
69.     INIT_LIST_HEAD(&pool->job_queue);
    
70.     INIT_LIST_HEAD(&pool->free_job_queue);
    
71. 
    
72.     if (pthread_mutex_init(&(pool->qlock), NULL))
    
73.     {
    
74.         printf("failed to init job queue mutex lock!\n");
    
75.         goto ERROR_OUT;
    
76.     }
    
77.     if (pthread_mutex_init(&(pool->fqlock), NULL))
    
78.     {
    
79.         printf("failed to init free job queue mutex lock!\n");
    
80.         goto ERROR_OUT1;
    
81.     }
    
82.     if (pthread_cond_init(&(pool->job_dequeue_valid), NULL))
    
83.     {
    
84.         printf("failed to init cond job_dequeue_valid!\n");
    
85.         goto ERROR_OUT2;
    
86.     }
    
87.     if (pthread_cond_init(&(pool->job_enqueue_valid), NULL))
    
88.     {
    
89.         printf("failed to init cond job_enqueue_valid!\n");
    
90.         goto ERROR_OUT3;
    
91.     }
    
92. 
    
93.     for (i = 0; i < max_job_num; i++)
    
94.     {
    
95.         job = (struct thread_job*)malloc(sizeof(struct thread_job));
    
96.         if (NULL == job)
    
97.         {
    
98.             goto ERROR_OUT4;
    
99.         }
    
100.         memset(job, 0, sizeof(struct thread_job));
     
101.         list_add(&job->list, &pool->free_job_queue);
     
102.     }
     
103. 
     
104.     pool->pthreads = malloc(sizeof(pthread_t) * thread_num);
     
105.     if (NULL == pool->pthreads)
     
106.     {
     
107.         printf("failed to malloc pthreads!\n");
     
108.         goto ERROR_OUT4;
     
109.     }
     
110.     memset(pool->pthreads, 0, sizeof(pthread_t) * thread_num);
     
111. 
     
112. 
     
113.     for (i = 0; i < pool->thread_num; ++i)
     
114.     {
     
115.         pthread_create(&(pool->pthreads[i]), NULL, thrdp_process_func, (void*)pool);
     
116.     }
     
117. 
     
118.     printf("create pool done\n");
     
119.     return pool;
     
120. 
     
121. ERROR_OUT4:
     
122.     if (!list_empty(&pool->free_job_queue))
     
123.     {
     
124.         list_for_each_entry_safe(job, tmp, &pool->free_job_queue, list)
     
125.         {
     
126.             list_del(&job->list);
     
127.             free(job);
     
128.         }
     
129.     }
     
130.     pthread_cond_destroy(&(pool->job_enqueue_valid));
     
131. ERROR_OUT3:
     
132.     pthread_cond_destroy(&(pool->job_dequeue_valid));
     
133. ERROR_OUT2:
     
134.     pthread_mutex_destroy(&(pool->fqlock));
     
135. ERROR_OUT1:
     
136.     pthread_mutex_destroy(&(pool->qlock));
     
137. ERROR_OUT:
     
138.     free(pool);
     
139.     return NULL;
     
140. 
     
141. }
     
142. 
     
143. void THRDP_destroy(struct thread_pool* pool)
     
144. {
     
145.     struct thread_job* job;
     
146.     struct thread_job* tmp;
     
147.     int i;
     
148. 
     
149.     if (!pool)
     
150.     {
     
151.         return;
     
152.     }
     
153.     pthread_mutex_lock(&(pool->qlock));
     
154.     if (thrdp_stopping(pool)) //线程池已经退出了，就直接返回
     
155.     {
     
156.         pthread_mutex_unlock(&(pool->qlock));
     
157.         return;
     
158.     }
     
159. 
     
160.     // 設置stop標記，喚醒阻塞的線程，使其退出
     
161.     pool->flags |= THRDP_STATUS_STOPPING;
     
162.     pthread_mutex_unlock(&(pool->qlock));
     
163. 
     
164.     pthread_cond_broadcast(&(pool->job_dequeue_valid)); //唤醒线程池中正在阻塞的线程
     
165.     pthread_cond_broadcast(&(pool->job_enqueue_valid)); //唤醒添加任务的threadpool_add_job函数
     
166. 
     
167.     for (i = 0; i < pool->thread_num; ++i)
     
168.     {
     
169.         pthread_join(pool->pthreads[i], NULL); //等待线程池的所有线程执行完毕
     
170.     }
     
171.     free(pool->pthreads);
     
172. 
     
173.     // all threads had done, no one using jobs, just free them
     
174.     if (!list_empty(&pool->job_queue))
     
175.     {
     
176.         list_for_each_entry_safe(job, tmp, &pool->job_queue, list)
     
177.         {
     
178.             list_del(&job->list);
     
179.             free(job);
     
180.         }
     
181.     }
     
182. 
     
183.     if (!list_empty(&pool->free_job_queue))
     
184.     {
     
185.         list_for_each_entry_safe(job, tmp, &pool->free_job_queue, list)
     
186.         {
     
187.             list_del(&job->list);
     
188.             free(job);
     
189.         }
     
190.     }
     
191. 
     
192.     pthread_mutex_destroy(&(pool->qlock));
     
193.     pthread_mutex_destroy(&(pool->fqlock));
     
194.     pthread_cond_destroy(&(pool->job_dequeue_valid));
     
195.     pthread_cond_destroy(&(pool->job_enqueue_valid));
     
196. }
     
197. 
     
198. /**
     
199. * get a free JOB
     
200. * if all jobs busy, wait util someone be free
     
201. */
     
202. struct thread_job* THRDP_job_get(struct thread_pool* pool)
     
203. {
     
204.     struct thread_job* job;
     
205. 
     
206.     if (!pool || thrdp_stopping(pool))
     
207.     {
     
208.         return NULL;
     
209.     }
     
210. 
     
211.     pthread_mutex_lock(&(pool->fqlock));
     
212.     while (list_empty(&pool->free_job_queue))
     
213.     {
     
214.         pthread_cond_wait(&(pool->job_enqueue_valid), &(pool->fqlock));
     
215.         if (thrdp_stopping(pool))
     
216.         {
     
217.             pthread_mutex_unlock(&(pool->fqlock));
     
218.             return NULL;
     
219.         }
     
220. 
     
221.     }
     
222.     job = list_first_entry(&pool->free_job_queue, struct thread_job, list);
     
223.     list_del(&job->list);
     
224.     pthread_mutex_unlock(&(pool->fqlock));
     
225.     job->pool = pool;
     
226.     return job;
     
227. }
     
228. 
     
229. 
     
230. /**
     
231. * put back a used job back to free_job_queue
     
232. *
     
233. * this will free the job data which MUST allocated dynamically
     
234. */
     
235. void THRDP_job_put(struct thread_pool* pool, struct thread_job* job)
     
236. {
     
237.     int notify = 0;
     
238.     if (!pool || !job )
     
239.     {
     
240.         return;
     
241.     }
     
242.     if (job->data)
     
243.     {
     
244.         free(job->data);
     
245.         job->data = NULL;
     
246.     }
     
247. 
     
248.     memset(job, 0, sizeof(struct thread_job));
     
249. 
     
250.     pthread_mutex_lock(&(pool->fqlock));
     
251.     notify = list_empty(&pool->free_job_queue);
     
252.     list_add_tail(&job->list, &pool->free_job_queue);
     
253.     pthread_mutex_unlock(&(pool->fqlock));
     
254. 
     
255.     if (notify)
     
256.     {
     
257.         printf("put back %p to free queue, notify wait jos tasks\n", job);
     
258.         pthread_cond_broadcast(&(pool->job_enqueue_valid));
     
259.     }
     
260. }
     
261. 
     
262. /**
     
263. * add a job to thread-pool for schedule
     
264. */
     
265. void THRDP_job_add(struct thread_pool* pool, struct thread_job* job)
     
266. {
     
267.     int notify = 0;
     
268. 
     
269.     if (!pool || !job )
     
270.     {
     
271.         return;
     
272.     }
     
273. 
     
274.     pthread_mutex_lock(&(pool->qlock));
     
275.     notify = list_empty(&pool->job_queue);
     
276.     list_add_tail(&job->list, &pool->job_queue);
     
277.     pthread_mutex_unlock(&(pool->qlock));
     
278. 
     
279.     if (notify)
     
280.     {
     
281.         // notify somebody to handle the new job
     
282.         printf("add new job, notify someone to handle it\n");
     
283.         pthread_cond_broadcast(&(pool->job_dequeue_valid));
     
284.     }
     
285. }

测试文件thrdp_test.c

点击(此处)折叠或打开

1. #include <stdio.h>
   
2. 
   
3. #include <string.h>
   
4. #include <stdlib.h>
   
5. #include <syslog.h>
   
6. #include <sys/socket.h>
   
7. #include <sys/types.h>
   
8. #include <signal.h>
   
9. #include <getopt.h>
   
10. #include <netinet/in.h>
    
11. #include <arpa/inet.h>
    
12. #include <string.h>
    
13. #include <sys/select.h>
    
14. #include <sys/un.h>
    
15. #include <stddef.h>
    
16. #include <unistd.h>
    
17. #include <errno.h>
    
18. #include <sys/stat.h>
    
19. #include <fcntl.h>
    
20. #include <sys/wait.h>
    
21. #include <unistd.h>
    
22. #include <uuid/uuid.h>
    
23. #include <sys/time.h>
    
24. 
    
25. #include "threadpool.h"
    
26. 
    
27. #define SRV_PORT 9966
    
28. #define UDP_PKG_MAX_LEN 4096
    
29. 
    
30. struct pkg_desc
    
31. {
    
32.     int srvfd;
    
33.     struct sockaddr_in cliaddr;
    
34.     char data[UDP_PKG_MAX_LEN];
    
35. };
    
36. 
    
37. 
    
38. void pkg_process(void* job)
    
39. {
    
40.     struct thread_job* pjob = (struct thread_job*) job;
    
41.     struct pkg_desc* ppkg = (struct pkg_desc*) pjob->data;
    
42. 
    
43.     printf("************* process in %u ******************\n", (unsigned int)pthread_self());
    
44.     printf("content:: %s\n", ppkg->data);
    
45.     sendto(ppkg->srvfd, ppkg->data, strlen(ppkg->data), 0, (struct sockaddr*)&ppkg->cliaddr, sizeof(ppkg->cliaddr));
    
46.     THRDP_job_put(pjob->pool, pjob);
    
47. }
    
48. 
    
49. 
    
50. void start_server(void)
    
51. {
    
52.     struct sockaddr_in srvin, cliin;
    
53.     socklen_t addrlen;
    
54.     int cmd_mgmt_fd;
    
55.     struct thread_pool* pool = NULL;
    
56. 
    
57. 
    
58.     cmd_mgmt_fd = socket(AF_INET, SOCK_DGRAM, 0);
    
59.     if(cmd_mgmt_fd < 0)
    
60.     {
    
61.         printf("start_cmd_mgmt_process socket fd failed.\n");
    
62.         return;
    
63.     }
    
64. 
    
65.     memset(&srvin, 0, sizeof(struct sockaddr_in));
    
66.     srvin.sin_family = AF_INET;
    
67.     srvin.sin_addr.s_addr = inet_addr("127.0.0.1");
    
68.     srvin.sin_port = htons(SRV_PORT);
    
69. 
    
70.     if(bind(cmd_mgmt_fd, (struct sockaddr*)&srvin, sizeof(struct sockaddr_in)) < 0)
    
71.     {
    
72.         printf("bind error");
    
73.     }
    
74. 
    
75. 
    
76.     if ((pool = THRDP_new(2, 8)) == NULL)
    
77.     {
    
78.         printf("create thread pool error\n");
    
79.         return;
    
80.     }
    
81. 
    
82.     while(1)
    
83.     {
    
84.         struct pkg_desc* ppkg = NULL;
    
85.         struct thread_job* job = THRDP_job_get(pool);
    
86.         int count;
    
87. 
    
88.         if (NULL == job)
    
89.         {
    
90.             printf("get thread process task error\n");
    
91.             return;
    
92.         }
    
93.         ppkg = (struct pkg_desc*) malloc(sizeof(struct pkg_desc));
    
94. 
    
95.         if (NULL == ppkg)
    
96.         {
    
97.             printf("malloc pkg error\n");
    
98.             return;
    
99.         }
    
100.         memset(ppkg, 0, sizeof(struct pkg_desc));
     
101. 
     
102. 
     
103.         addrlen = sizeof(struct sockaddr_in);
     
104.         count = recvfrom(cmd_mgmt_fd, ppkg->data, UDP_PKG_MAX_LEN, 0, (struct sockaddr*)&ppkg->cliaddr, &addrlen);
     
105.         if(count <= 0)
     
106.         {
     
107.             printf("recvform error\n");
     
108.             continue;
     
109.         }
     
110. 
     
111.         ppkg->srvfd = cmd_mgmt_fd;
     
112.         job->data = ppkg;
     
113.         job->jfunc = pkg_process;
     
114.         THRDP_job_add(pool, job);
     
115.     }
     
116. }
     
117. 
     
118. 
     
119. void start_client(int client_id)
     
120. {
     
121.     struct sockaddr_in srvaddr;
     
122.     struct sockaddr_in peeraddr;
     
123.     int len = sizeof(peeraddr);
     
124.     int fd;
     
125.     char ctnbuf[512] = {0};
     
126.     char rcvbuf[512] = {0};
     
127.     unsigned int i = 0;
     
128.     fd = socket(AF_INET, SOCK_DGRAM, 0);
     
129.     if(fd < 0)
     
130.     {
     
131.         printf("start_client socket fd failed.\n");
     
132.         return;
     
133.     }
     
134. 
     
135.     memset(&srvaddr, 0, sizeof(struct sockaddr_in));
     
136.     srvaddr.sin_family = AF_INET;
     
137.     srvaddr.sin_addr.s_addr = inet_addr("127.0.0.1");
     
138.     srvaddr.sin_port = htons(SRV_PORT);
     
139. 
     
140. 
     
141.     while(1)
     
142.     {
     
143. 
     
144.         memset(ctnbuf, 0, 512);
     
145.         sprintf(ctnbuf, "%d say just test:%u", client_id, i++);
     
146. 
     
147.         sendto(fd, ctnbuf, strlen(ctnbuf), 0, (struct sockaddr*)&srvaddr, sizeof(srvaddr));
     
148.         memset(rcvbuf, 0, 512);
     
149.         len = sizeof(peeraddr);
     
150.         recvfrom(fd, rcvbuf, 512, 0, (struct sockaddr*)&peeraddr, &len);
     
151.         printf("recv content:%s\n", rcvbuf);
     
152.     }
     
153. }
     
154. 
     
155. int main(int argc, char** argv)
     
156. {
     
157.     if (argc < 2)
     
158.     {
     
159.         printf("usage : %s [-s/-c] [cid]", argv[0]);
     
160.         return 0;
     
161.     }
     
162.     if (!strcmp(argv[1], "-s"))
     
163.     {
     
164.         start_server();
     
165.     }
     
166.     else
     
167.     {
     
168.         start_client(atoi(argv[2]));
     
169.     }
     
170.     return 0;
     
171. 
     
172. }

测试是一个简单的udp回射服务器模型，用线程池的方式做服务端并发。

编译： gcc -g -o thrdp *.c

启动server端： ./thrdp -s

多终端启动client端： ./thrdp -c 100，./thrdp -c 102， ...

运行截图如下