有時我們會需要大量線程來處理一些相互獨立的任務，為了避免頻繁的申請釋放線程所帶來的開銷，我們可以使用線程池。下面是一個C語言實現的簡單的線程池。
頭文件：
1: #ifndef THREAD_POOL_H__
2: #define THREAD_POOL_H__
3:
4: #include <pthread.h>
5:
6: /* 要執行的任務鏈表 */
7: typedef struct tpool_work {
8: void* (*routine)(void*); /* 任務函數 */
9: void *arg; /* 傳入任務函數的參數 */
10: struct tpool_work *next;
11: }tpool_work_t;
12:
13: typedef struct tpool {
14: int shutdown; /* 線程池是否銷毀 */
15: int max_thr_num; /* 最大線程數 */
16: pthread_t *thr_id; /* 線程ID數組 */
17: tpool_work_t *queue_head; /* 線程鏈表 */
18: pthread_mutex_t queue_lock;
19: pthread_cond_t queue_ready;
20: }tpool_t;
21:
22: /*
23: * @brief 創建線程池
24: * @param max_thr_num 最大線程數
25: * @return 0: 成功 其他: 失敗
26: */
27: int
28: tpool_create(int max_thr_num);
29:
30: /*
31: * @brief 銷毀線程池
32: */
33: void
34: tpool_destroy();
35:
36: /*
37: * @brief 向線程池中添加任務
38: * @param routine 任務函數指針
39: * @param arg 任務函數參數
40: * @return 0: 成功 其他:失敗
41: */
42: int
43: tpool_add_work(void*(*routine)(void*), void *arg);
44:
45: #endif
實現:
1: #include <unistd.h>
2: #include <stdlib.h>
3: #include <errno.h>
4: #include <string.h>
5: #include <stdio.h>
6:
7: #include "tpool.h"
8:
9: static tpool_t *tpool = NULL;
10:
11: /* 工作者線程函數, 從任務鏈表中取出任務並執行 */
12: static void*
13: thread_routine(void *arg)
14: {
15: tpool_work_t *work;
16:
17: while(1) {
18: /* 如果線程池沒有被銷毀且沒有任務要執行，則等待 */
19: pthread_mutex_lock(&tpool->queue_lock);
20: while(!tpool->queue_head && !tpool->shutdown) {
21: pthread_cond_wait(&tpool->queue_ready, &tpool->queue_lock);
22: }
23: if (tpool->shutdown) {
24: pthread_mutex_unlock(&tpool->queue_lock);
25: pthread_exit(NULL);
26: }
27: work = tpool->queue_head;
28: tpool->queue_head = tpool->queue_head->next;
29: pthread_mutex_unlock(&tpool->queue_lock);
30:
31: work->routine(work->arg);
32: free(work);
33: }
34:
35: return NULL;
36: }
37:
38: /*
39: * 創建線程池
40: */
41: int
42: tpool_create(int max_thr_num)
43: {
44: int i;
45:
46: tpool = calloc(1, sizeof(tpool_t));
47: if (!tpool) {
48: printf("%s: calloc failed\n", __FUNCTION__);
49: exit(1);
50: }
51:
52: /* 初始化 */
53: tpool->max_thr_num = max_thr_num;
54: tpool->shutdown = 0;
55: tpool->queue_head = NULL;
56: if (pthread_mutex_init(&tpool->queue_lock, NULL) !=0) {
57: printf("%s: pthread_mutex_init failed, errno:%d, error:%s\n",
58: __FUNCTION__, errno, strerror(errno));
59: exit(1);
60: }
61: if (pthread_cond_init(&tpool->queue_ready, NULL) !=0 ) {
62: printf("%s: pthread_cond_init failed, errno:%d, error:%s\n",
63: __FUNCTION__, errno, strerror(errno));
64: exit(1);
65: }
66:
67: /* 創建工作者線程 */
68: tpool->thr_id = calloc(max_thr_num, sizeof(pthread_t));
69: if (!tpool->thr_id) {
70: printf("%s: calloc failed\n", __FUNCTION__);
71: exit(1);
72: }
73: for (i = 0; i < max_thr_num; ++i) {
74: if (pthread_create(&tpool->thr_id[i], NULL, thread_routine, NULL) != 0){
75: printf("%s:pthread_create failed, errno:%d, error:%s\n", __FUNCTION__,
76: errno, strerror(errno));
77: exit(1);
78: }
79:
80: }
81:
82: return 0;
83: }
84:
85: /* 銷毀線程池 */
86: void
87: tpool_destroy()
88: {
89: int i;
90: tpool_work_t *member;
91:
92: if (tpool->shutdown) {
93: return;
94: }
95: tpool->shutdown = 1;
96:
97: /* 通知所有正在等待的線程 */
98: pthread_mutex_lock(&tpool->queue_lock);
99: pthread_cond_broadcast(&tpool->queue_ready);
100: pthread_mutex_unlock(&tpool->queue_lock);
101: for (i = 0; i < tpool->max_thr_num; ++i) {
102: pthread_join(tpool->thr_id[i], NULL);
103: }
104: free(tpool->thr_id);
105:
106: while(tpool->queue_head) {
107: member = tpool->queue_head;
108: tpool->queue_head = tpool->queue_head->next;
109: free(member);
110: }
111:
112: pthread_mutex_destroy(&tpool->queue_lock);
113: pthread_cond_destroy(&tpool->queue_ready);
114:
115: free(tpool);
116: }
117:
118: /* 向線程池添加任務 */
119: int
120: tpool_add_work(void*(*routine)(void*), void *arg)
121: {
122: tpool_work_t *work, *member;
123:
124: if (!routine){
125: printf("%s:Invalid argument\n", __FUNCTION__);
126: return -1;
127: }
128:
129: work = malloc(sizeof(tpool_work_t));
130: if (!work) {
131: printf("%s:malloc failed\n", __FUNCTION__);
132: return -1;
133: }
134: work->routine = routine;
135: work->arg = arg;
136: work->next = NULL;
137:
138: pthread_mutex_lock(&tpool->queue_lock);
139: member = tpool->queue_head;
140: if (!member) {
141: tpool->queue_head = work;
142: } else {
143: while(member->next) {
144: member = member->next;
145: }
146: member->next = work;
147: }
148: /* 通知工作者線程，有新任務添加 */
149: pthread_cond_signal(&tpool->queue_ready);
150: pthread_mutex_unlock(&tpool->queue_lock);
151:
152: return 0;
153: }
154:
155:
測試代碼:
1: #include <unistd.h>
2: #include <stdio.h>
3: #include <stdlib.h>
4: #include "tpool.h"
5:
6: void *func(void *arg)
7: {
8: printf("thread %d\n", (int)arg);
9: return NULL;
10: }
11:
12: int
13: main(int arg, char **argv)
14: {
15: if (tpool_create(5) != 0) {
16: printf("tpool_create failed\n");
17: exit(1);
18: }
19:
20: int i;
21: for (i = 0; i < 10; ++i) {
22: tpool_add_work(func, (void*)i);
23: }
24: sleep(2);
25: tpool_destroy();
26: return 0;
27: }
這個實現是在調用tpool_destroy之後，僅將當前正在執行的任務完成之後就會退出，我們也可以修改代碼使得線程池在執行完任務鏈表中所有任務後再退出。