Linux线程池(C语言描述) - 互斥量+条件变量同步

今天药忘吃喽~ 2022-08-10 05:46 128阅读 0赞

创建线程或者进程的开销是很大的，为了防止频繁的创建线程，提高程序的运行效率，往往都会建立一个线程池用于多线程程序的调度  
  
下面的程序就是完整的线程池实现，主要采用互斥量和条件变量实现同步  
   
首先定义头文件threadpool.h  
  
在该文件中定义了线程池的数据结构和所有的函数

**\[cpp\]**  [view plain][] [copy][view plain]

1.  \#ifndef THREADPOOL\_H\_  
2.  \#define THREADPOOL\_H\_  
3.    
4.  \#include <stdio.h>  
5.  \#include <stdlib.h>  
6.  \#include <unistd.h>  
7.  \#include <pthread.h>  
8.    
9.  /\*\* 
10.  \* 线程体数据结构 
11.  \*/  
12. **typedef** **struct** runner  
13. \{  
14.  **void** (\*callback)(**void**\* arg); // 回调函数指针  
15.  **void**\* arg; // 回调函数的参数  
16.  **struct** runner\* next;  
17. \} thread\_runner;  
18.   
19. /\*\* 
20.  \* 线程池数据结构 
21.  \*/  
22. **typedef** **struct**  
23. \{  
24.  pthread\_mutex\_t mutex; //互斥量  
25.  pthread\_cond\_t cond; // 条件变量  
26.  thread\_runner\* runner\_head; // 线程池中所有等待任务的头指针  
27.  thread\_runner\* runner\_tail; // 线程池所有等待任务的尾指针  
28.  **int** shutdown; // 线程池是否销毁  
29.  pthread\_t\* threads; // 所有线程  
30.  **int** max\_thread\_size; //线程池中允许的活动线程数目  
31. \} thread\_pool;  
32.   
33. /\*\* 
34.  \* 线程体 
35.  \*/  
36. **void** run(**void** \*arg);  
37.   
38. /\*\* 
39.  \* 初始化线程池 
40.  \*  参数： 
41.  \*   pool：指向线程池结构有效地址的动态指针 
42.  \*   max\_thread\_size：最大的线程数 
43.  \*/  
44. **void** threadpool\_init(thread\_pool\* pool, **int** max\_thread\_size);  
45.   
46. /\*\* 
47.  \* 向线程池加入任务 
48.  \*  参数： 
49.  \*   pool：指向线程池结构有效地址的动态指针 
50.  \*   callback：线程回调函数 
51.  \*   arg：回调函数参数 
52.  \*/  
53. **void** threadpool\_add\_runner(thread\_pool\* pool, **void** (\*callback)(**void** \*arg), **void** \*arg);  
54.   
55. /\*\* 
56.  \* 销毁线程池 
57.  \*  参数： 
58.  \*   ppool：指向线程池结构有效地址的动态指针地址(二级指针)，销毁后释放内存，该指针为NULL 
59.  \*/  
60. **void** threadpool\_destroy(thread\_pool\*\* ppool);  
61.   
62. \#endif

然后是函数实现threadpool.h  
  
  
该文件实现了threadpool.h的函数定义

**\[cpp\]**  [view plain][] [copy][view plain]

1.  \#include "threadpool.h"  
2.    
3.  \#define DEBUG 1  
4.    
5.  /\*\* 
6.   \* 线程体 
7.   \*/  
8.  **void** run(**void** \*arg)  
9.  \{  
10.  thread\_pool\* pool = (thread\_pool\*) arg;  
11.  **while** (1)  
12.  \{  
13.   // 加锁  
14.   pthread\_mutex\_lock(&(pool->mutex));  
15. \#ifdef DEBUG  
16.   printf("run-> locked\\n");  
17. \#endif  
18.   // 如果等待队列为0并且线程池未销毁，则处于阻塞状态  
19.   **while** (pool->runner\_head == NULL && !pool->shutdown)  
20.   \{  
21.    pthread\_cond\_wait(&(pool->cond), &(pool->mutex));  
22.   \}  
23.   //如果线程池已经销毁  
24.   **if** (pool->shutdown)  
25.   \{  
26.    // 解锁  
27.    pthread\_mutex\_unlock(&(pool->mutex));  
28. \#ifdef DEBUG  
29.    printf("run-> unlocked and thread exit\\n");  
30. \#endif  
31.    pthread\_exit(NULL);  
32.   \}  
33.   // 取出链表中的头元素  
34.   thread\_runner \*runner = pool->runner\_head;  
35.   pool->runner\_head = runner->next;  
36.   // 解锁  
37.   pthread\_mutex\_unlock(&(pool->mutex));  
38. \#ifdef DEBUG  
39.   printf("run-> unlocked\\n");  
40. \#endif  
41.   // 调用回调函数，执行任务  
42.   (runner->callback)(runner->arg);  
43.   free(runner);  
44.   runner = NULL;  
45. \#ifdef DEBUG  
46.   printf("run-> runned and free runner\\n");  
47. \#endif  
48.  \}  
49.  pthread\_exit(NULL);  
50. \}  
51.   
52. /\*\* 
53.  \* 初始化线程池 
54.  \*  参数： 
55.  \*   pool：指向线程池结构有效地址的动态指针 
56.  \*   max\_thread\_size：最大的线程数 
57.  \*/  
58. **void** threadpool\_init(thread\_pool\* pool, **int** max\_thread\_size)  
59. \{  
60.  // 初始化互斥量  
61.  pthread\_mutex\_init(&(pool->mutex), NULL);  
62.  // 初始化条件变量  
63.  pthread\_cond\_init(&(pool->cond), NULL);  
64.  pool->runner\_head = NULL;  
65.  pool->runner\_tail = NULL;  
66.  pool->max\_thread\_size = max\_thread\_size;  
67.  pool->shutdown = 0;  
68.  // 创建所有分离态线程  
69.  pool->threads = (pthread\_t \*) malloc(max\_thread\_size \* **sizeof**(pthread\_t));  
70.  **int** i = 0;  
71.  **for** (i = 0; i < max\_thread\_size; i++)  
72.  \{  
73.   pthread\_attr\_t attr;  
74.   pthread\_attr\_init(&attr);  
75.   pthread\_attr\_setdetachstate(&attr, PTHREAD\_CREATE\_DETACHED);  
76.   pthread\_create(&(pool->threads\[i\]), &attr, (**void**\*) run, (**void**\*) pool);  
77.  \}  
78. \#ifdef DEBUG  
79.  printf("threadpool\_init-> create %d detached thread\\n", max\_thread\_size);  
80. \#endif  
81. \}  
82.   
83. /\*\* 
84.  \* 向线程池加入任务 
85.  \*  参数： 
86.  \*   pool：指向线程池结构有效地址的动态指针 
87.  \*   callback：线程回调函数 
88.  \*   arg：回调函数参数 
89.  \*/  
90. **void** threadpool\_add\_runner(thread\_pool\* pool, **void** (\*callback)(**void** \*arg), **void** \*arg)  
91. \{  
92.  // 构造一个新任务  
93.  thread\_runner \*newrunner = (thread\_runner \*) malloc(**sizeof**(thread\_runner));  
94.  newrunner->callback = callback;  
95.  newrunner->arg = arg;  
96.  newrunner->next = NULL;  
97.  // 加锁  
98.  pthread\_mutex\_lock(&(pool->mutex));  
99. \#ifdef DEBUG  
100.  printf("threadpool\_add\_runner-> locked\\n");  
101. \#endif  
102.  // 将任务加入到等待队列中  
103.  **if** (pool->runner\_head != NULL)  
104.  \{  
105.   pool->runner\_tail->next = newrunner;  
106.   pool->runner\_tail = newrunner;  
107.  \}  
108.  **else**  
109.  \{  
110.   pool->runner\_head = newrunner;  
111.   pool->runner\_tail = newrunner;  
112.  \}  
113.  // 解锁  
114.  pthread\_mutex\_unlock(&(pool->mutex));  
115. \#ifdef DEBUG  
116.  printf("threadpool\_add\_runner-> unlocked\\n");  
117. \#endif  
118.  // 唤醒一个等待线程  
119.  pthread\_cond\_signal(&(pool->cond));  
120. \#ifdef DEBUG  
121.  printf("threadpool\_add\_runner-> add a runner and wakeup a waiting thread\\n");  
122. \#endif  
123. \}  
124.   
125. /\*\* 
126.  \* 销毁线程池 
127.  \*  参数： 
128.  \*   ppool：指向线程池结构有效地址的动态指针地址(二级指针) 
129.  \*/  
130. **void** threadpool\_destroy(thread\_pool\*\* ppool)  
131. \{  
132.  thread\_pool \*pool = \*ppool;  
133.  // 防止2次销毁  
134.  **if** (!pool->shutdown)  
135.  \{  
136.   pool->shutdown = 1;  
137.   // 唤醒所有等待线程，线程池要销毁了  
138.   pthread\_cond\_broadcast(&(pool->cond));  
139.   // 等待所有线程中止  
140.   sleep(1);  
141. \#ifdef DEBUG  
142.   printf("threadpool\_destroy-> wakeup all waiting threads\\n");  
143. \#endif  
144.   // 回收空间  
145.   free(pool->threads);  
146.   // 销毁等待队列  
147.   thread\_runner \*head = NULL;  
148.   **while** (pool->runner\_head != NULL)  
149.   \{  
150.    head = pool->runner\_head;  
151.    pool->runner\_head = pool->runner\_head->next;  
152.    free(head);  
153.   \}  
154.   
155. \#ifdef DEBUG  
156.   printf("threadpool\_destroy-> all runners freed\\n");  
157. \#endif  
158.   /\*条件变量和互斥量也别忘了销毁\*/  
159.   pthread\_mutex\_destroy(&(pool->mutex));  
160.   pthread\_cond\_destroy(&(pool->cond));  
161.   
162. \#ifdef DEBUG  
163.   printf("threadpool\_destroy-> mutex and cond destoryed\\n");  
164. \#endif  
165.   free(pool);  
166.   (\*ppool) = NULL;  
167.   
168. \#ifdef DEBUG  
169.   printf("threadpool\_destroy-> pool freed\\n");  
170. \#endif  
171.  \}  
172. \}

以上就是完整的线城池实现  
  
  
下面写个测试程序来看看

**\[cpp\]**  [view plain][] [copy][view plain]

1.  \#include "threadpool.h"  
2.    
3.  **void** threadrun(**void**\* arg)  
4.  \{  
5.   **int** \*i = (**int** \*) arg;  
6.   printf("%d\\n", \*i);  
7.  \}  
8.    
9.  **int** main(**void**)  
10. \{  
11.  thread\_pool \*pool = malloc(**sizeof**(thread\_pool));  
12.  threadpool\_init(pool, 2);  
13.  **int** i;  
14.  **int** tmp\[3\];  
15.  **for** (i = 0; i < 3; i++)  
16.  \{  
17.   tmp\[i\] = i;  
18.   threadpool\_add\_runner(pool, threadrun, &tmp\[i\]);  
19.  \}  
20.  sleep(1);  
21.  threadpool\_destroy(&pool);  
22.  printf("main-> %p\\n",pool);  
23.  printf("main-> test over\\n");  
24.  **return** 0;  
25. \}

该函数创建了1个线程池，包含2个线程，然后往线程池中加入了3个任务  
  
并在最后打印了线程池的指针地址(主要是为了看线程池销毁后的状态)  
  
运行结果如下:

![2726085149460144979.jpg][]

输出的信息主要调试信息，  
值得注意的是倒数第二行“main-> (nil)”  
说明线程池销毁后，指向线程池的指针 = NULL（这就是为什么在threadpool\_destory函数为什么要用二级指针的原因）

[view plain]: http://blog.csdn.net/gzshun/article/details/8178067#
[2726085149460144979.jpg]: http://img9.ph.126.net/xYI8i4cTedBxYnlT4OL7UQ==/2726085149460144979.jpg