线程池

标签：

1. #include <stdio.h>   
2. #include <stdlib.h>   
3. #include <unistd.h>   
4. #include <sys/types.h>   
5. #include <pthread.h>   
6. #include <assert.h>   
7.   
8. /*  
9. *线程池里所有运行和等待的任务都是一个CThread_worker  
10. *由于所有任务都在链表里，所以是一个链表结构  
11. */   
12. typedef struct worker   
13. {   
14.     /*回调函数，任务运行时会调用此函数，注意也可声明成其它形式*/   
15.     void *(*process) (void *arg);   
16.     void *arg;/*回调函数的参数*/   
17.     struct worker *next;   
18.   
19. } CThread_worker;   
20.     
21. /*线程池结构*/   
22. typedef struct   
23. {   
24.     pthread_mutex_t queue_lock;   
25.     pthread_cond_t queue_ready;   
26.   
27.     /*链表结构，线程池中所有等待任务*/   
28.     CThread_worker *queue_head;   
29.   
30.     /*是否销毁线程池*/   
31.     int shutdown;   
32.     pthread_t *threadid;   
33.     /*线程池中允许的活动线程数目*/   
34.     int max_thread_num;   
35.     /*当前等待队列的任务数目*/   
36.     int cur_queue_size;   
37.   
38. } CThread_pool;   
39.     
40. int pool_add_worker (void *(*process) (void *arg), void *arg);   
41. void *thread_routine (void *arg);   
42.   
43. static CThread_pool *pool = NULL;   
44. void pool_init (int max_thread_num)   
45. {   
46.     pool = (CThread_pool *) malloc (sizeof (CThread_pool));   
47.   
48.     pthread_mutex_init (&(pool->queue_lock), NULL);   
49.     /*初始化条件变量*/  
50.     pthread_cond_init (&(pool->queue_ready), NULL);   
51.   
52.     pool->queue_head = NULL;   
53.   
54.     pool->max_thread_num = max_thread_num;   
55.     pool->cur_queue_size = 0;   
56.   
57.     pool->shutdown = 0;   
58.   
59.     pool->threadid =   
60.         (pthread_t *) malloc (max_thread_num * sizeof (pthread_t));   
61.     int i = 0;   
62.     for (i = 0; i < max_thread_num; i++)   
63.     {    
64.         pthread_create (&(pool->threadid[i]), NULL, thread_routine,   
65.                 NULL);   
66.     }   
67. }     
68. /*向线程池中加入任务*/   
69. int   
70. pool_add_worker (void *(*process) (void *arg), void *arg)   
71. {   
72.     /*构造一个新任务*/   
73.     CThread_worker *newworker =                   
74.         (CThread_worker *) malloc (sizeof (CThread_worker));    /*将任务传递进来*/  
75.     newworker->process = process;    /*任务函数*/  
76.     newworker->arg = arg;            /*函数参数*/  
77.     newworker->next = NULL;/*别忘置空*/   
78.   
79.     pthread_mutex_lock (&(pool->queue_lock));   
80.     /*将任务加入到等待队列中*/   
81.     CThread_worker *member = pool->queue_head;   
82.     if (member != NULL)   
83.     {   
84.         while (member->next != NULL)   
85.             member = member->next;   
86.         member->next = newworker;   
87.     }   
88.     else   
89.     {   
90.         pool->queue_head = newworker;   
91.     }   
92.   
93.     assert (pool->queue_head != NULL);   
94.   
95.     pool->cur_queue_size++;   
96.     pthread_mutex_unlock (&(pool->queue_lock));     
97.     pthread_cond_signal (&(pool->queue_ready));    /*每次等待队列中有任务加入，都会试图唤醒一个等待线程；   注意如果所有线程都在忙碌，这句没有任何作用*/
98.     return 0;   
99. }   
100.   
101. /*销毁线程池，等待队列中的任务不会再被执行，但是正在运行的线程会一直  
102. 把任务运行完后再退出*/   
103. int pool_destroy ()   
104. {   
105.     if (pool->shutdown)   
106.         return -1;/*防止两次调用*/   
107.     pool->shutdown = 1;   
108.   
109.     /*唤醒所有等待线程，线程池要销毁了*/   
110.     pthread_cond_broadcast (&(pool->queue_ready));   
111.   
112.     /*阻塞等待线程退出，否则就成僵尸了*/   
113.     int i;   
114.     for (i = 0; i < pool->max_thread_num; i++)   
115.         pthread_join (pool->threadid[i], NULL);   
116.     free (pool->threadid);   
117.   
118.     /*销毁等待队列*/   
119.     CThread_worker *head = NULL;   
120.     while (pool->queue_head != NULL)   
121.     {   
122.         head = pool->queue_head;   
123.         pool->queue_head = pool->queue_head->next;   
124.         free (head);   
125.     }   
126.     /*条件变量和互斥量也别忘了销毁*/   
127.     pthread_mutex_destroy(&(pool->queue_lock));   
128.     pthread_cond_destroy(&(pool->queue_ready));   
129.        
130.     free (pool);   
131.     /*销毁后指针置空是个好习惯*/   
132.     pool=NULL;   
133.     return 0;   
134. }    
135. void * thread_routine (void *arg)       /*线程调用函数*/   
136. {   
137.     printf ("starting thread 0x%x\n", pthread_self ());   
138.     while (1)   
139.     {   
140.         pthread_mutex_lock (&(pool->queue_lock));   
141.         /*如果等待队列为0并且不销毁线程池，则处于阻塞状态; 注意  
142.         pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁*/   
143.         while (pool->cur_queue_size == 0 && !pool->shutdown)   
144.         {   
145.             printf ("thread 0x%x is waiting\n", pthread_self ());   
146.             pthread_cond_wait (&(pool->queue_ready), &(pool->queue_lock));   
147.         }   
148.   
149.         /*线程池要销毁了*/   
150.         if (pool->shutdown)   
151.         {   
152.             /*遇到break,continue,return等跳转语句，千万不要忘记先解锁*/   
153.             pthread_mutex_unlock (&(pool->queue_lock));   
154.             printf ("thread 0x%x will exit\n", pthread_self ());   
155.             pthread_exit (NULL);   
156.         }   
157.   
158.         printf ("thread 0x%x is starting to work\n", pthread_self ());   
159.   
160.         /*assert是调试的好帮手*/   
161.         assert (pool->cur_queue_size != 0);   
162.         assert (pool->queue_head != NULL);   
163.            
164.         /*等待队列长度减去1，并取出链表中的头元素*/   
165.         pool->cur_queue_size--;   
166.         CThread_worker *worker = pool->queue_head;   
167.         pool->queue_head = worker->next;   
168.         pthread_mutex_unlock (&(pool->queue_lock));   
169.   
170.         /*调用回调函数，执行任务*/   
171.         (*(worker->process)) (worker->arg);   
172.         free (worker);   
173.         worker = NULL;   
174.     }   
175.     /*这一句应该是不可达的*/   
176.     pthread_exit (NULL);   
177. }  
178.   
179. void * myprocess (void *arg)   
180. {   
181.     printf ("threadid is 0x%x, working on task %d\n", pthread_self (),*(int *) arg);   
182.     sleep (1);/*休息一秒，延长任务的执行时间*/   
183.     return NULL;   
184. }   
185.   
186. int main (int argc, char **argv)   
187. {   
188.     pool_init (3);/*线程池中最多三个活动线程*/   
189.        
190.     /*连续向池中投入10个任务*/   
191.     int *workingnum = (int *) malloc (sizeof (int) * 10);       /*任务编号*/  
192.     int i;   
193.     for (i = 0; i < 10; i++)   
194.     {   
195.         workingnum[i] = i;   
196.         pool_add_worker (myprocess, &workingnum[i]);    /*任务添加进等待队列*/  
197.     }   
198.     /*等待所有任务完成*/   
199.     sleep (5);   
200.     /*销毁线程池*/   
201.     pool_destroy ();   
202.   
203.     free (workingnum);   
204.     return 0; 

线程池

标签：

原文地址：http://www.cnblogs.com/liuchengchuxiao/p/4329913.html