线程池的简单实现

  1 #include <stdio.h>
  2 #include <stdlib.h>
  3 #include <pthread.h>    //linux环境中多线程的头文件，非C语言标准库，编译时最后要加 -lpthread 调用动态链接库
  4 
  5 //工作链表的结构
  6 typedef struct worker {
  7     void *(*process) (void *arg);   //工作函数
  8     void *arg;                      //函数的参数
  9     struct worker *next;
 10 }CThread_worker;
 11 
 12 //线程池的结构
 13 typedef struct {
 14     pthread_mutex_t queue_lock;     //互斥锁
 15     pthread_cond_t  queue_ready;    //条件变量/信号量
 16 
 17     CThread_worker *queue_head;     //指向工作链表的头结点，临界区
 18     int cur_queue_size;             //记录链表中工作的数量，临界区
 19 
 20     int max_thread_num;             //最大线程数
 21     pthread_t *threadid;            //线程ID
 22 
 23     int shutdown;                   //开关
 24 }CThread_pool;
 25 
 26 static CThread_pool *pool = NULL;   //一个线程池变量
 27 int pool_add_worker(void *(*process)(void *arg), void *arg);    //负责向工作链表中添加工作
 28 void *thread_routine(void *arg);    //线程例程
 29 
 30 //线程池初始化
 31 void
 32 pool_init(int max_thread_num)
 33 {
 34     int i = 0;
 35 
 36     pool = (CThread_pool *) malloc (sizeof (CThread_pool));    //创建线程池
 37 
 38     pthread_mutex_init(&(pool->queue_lock),  NULL);     //互斥锁初始化，参数为锁的地址
 39     pthread_cond_init( &(pool->queue_ready), NULL);     //条件变量初始化，参数为变量地址
 40 
 41     pool->queue_head = NULL;
 42     pool->cur_queue_size = 0;
 43 
 44     pool->max_thread_num = max_thread_num;
 45     pool->threadid = (pthread_t *) malloc(max_thread_num * sizeof(pthread_t));
 46     for (i = 0; i < max_thread_num; i++) {
 47         pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);  //创建线程, 参数为线程ID变量地址、属性、例程、参数
 48     }
 49 
 50     pool->shutdown = 0;
 51 }
 52 
 53 //例程，调用具体的工作函数
 54 void *
 55 thread_routine(void *arg)
 56 {
 57     printf("starting thread 0x%x\n", (int)pthread_self());
 58     while(1) {
 59         pthread_mutex_lock(&(pool->queue_lock));    //从工作链表中取工作，要先加互斥锁，参数为锁地址
 60 
 61         while(pool->cur_queue_size == 0 && !pool->shutdown) {       //链表为空
 62             printf("thread 0x%x is waiting\n", (int)pthread_self());
 63             pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));   //等待资源，信号量用于通知。会释放第二个参数的锁，以供添加；函数返回时重新加锁。
 64         }
 65 
 66         if(pool->shutdown) {
 67             pthread_mutex_unlock(&(pool->queue_lock));          //结束开关开启，释放锁并退出线程
 68             printf("thread 0x%x will exit\n", (int)pthread_self());
 69             pthread_exit(NULL);     //参数为void *
 70         }
 71 
 72         printf("thread 0x%x is starting to work\n", (int)pthread_self());
 73 
 74         --pool->cur_queue_size;
 75         CThread_worker *worker = pool->queue_head;
 76         pool->queue_head = worker->next;
 77 
 78         pthread_mutex_unlock (&(pool->queue_lock));     //获取一个工作后释放锁
 79 
 80 
 81         (*(worker->process))(worker->arg);      //做工作
 82         free(worker);
 83         worker = NULL;
 84     }
 85 
 86     pthread_exit(NULL);
 87 }
 88 
 89 //销毁线程池
 90 int
 91 pool_destroy()
 92 {
 93     if(pool->shutdown)      //检测结束开关是否开启，若开启，则所有线程会自动退出
 94         return -1;
 95     pool->shutdown = 1;
 96 
 97     pthread_cond_broadcast( &(pool->queue_ready) );     //广播，唤醒所有线程，准备退出
 98 
 99     int i;
100     for(i = 0; i < pool->max_thread_num; ++i)
101         pthread_join(pool->threadid[i], NULL);      //主线程等待所有线程退出，只有join第一个参数不是指针，第二个参数类型是void **，接收exit的返回值，需要强制转换
102     free(pool->threadid);
103 
104     CThread_worker *head = NULL;
105     while(pool->queue_head != NULL) {           //释放未执行的工作链表剩余结点
106         head = pool->queue_head;
107         pool->queue_head = pool->queue_head->next;
108         free(head);
109     }
110 
111     pthread_mutex_destroy(&(pool->queue_lock));     //销毁锁和条件变量
112     pthread_cond_destroy(&(pool->queue_ready));
113 
114     free(pool);
115     pool=NULL;
116     return 0;
117 }
118 
119 void *
120 myprocess(void *arg)
121 {
122     printf("threadid is 0x%x, working on task %d\n", (int)pthread_self(), *(int*)arg);
123     sleep (1);
124     return NULL;
125 }
126 
127 //添加工作
128 int
129 pool_add_worker(void *(*process)(void *arg), void *arg)
130 {
131     CThread_worker *newworker = (CThread_worker *) malloc(sizeof(CThread_worker));
132     newworker->process = process;   //具体的工作函数
133     newworker->arg = arg;
134     newworker->next = NULL;
135 
136     pthread_mutex_lock( &(pool->queue_lock) );      //加锁
137 
138     CThread_worker *member = pool->queue_head;      //插入链表尾部
139     if( member != NULL ) {
140         while( member->next != NULL )
141             member = member->next;
142         member->next = newworker;
143     }
144     else {
145         pool->queue_head = newworker;
146     }
147     ++pool->cur_queue_size;
148 
149     pthread_mutex_unlock( &(pool->queue_lock) );   //解锁
150 
151     pthread_cond_signal( &(pool->queue_ready) );   //通知一个等待的线程
152     return 0;
153 }
154 
155 int
156 main(int argc, char **argv)
157 {
158     pool_init(3);   //主线程创建线程池，3个线程
159 
160     int *workingnum = (int *) malloc(sizeof(int) * 10);
161     int i;
162     for(i = 0; i < 10; ++i) {
163         workingnum[i] = i;
164         pool_add_worker(myprocess, &workingnum[i]);     //主线程负责添加工作，10个工作
165     }
166 
167     sleep (5);
168     pool_destroy();     //销毁线程池
169     free (workingnum);
170 
171     return 0;
172 }