//c语言线程池实现#ifndef __THREADPOOL__#define __THREADPOOL__#ifdef __cplusplusextern "C" {#endif //#endif#include #include #include #include #include #include /**线程池里所有运行和等待的任务都是一个CThread_Worker*由于所有任务都在链表里，所以是一个链表结构*/typedef struct Worker{ /*回调函数，任务运行时会调用此函数*/ void (*process)(void *arg); void *arg;/*回调函数的参数*/ struct Worker *next; }CThreadWorker;/*线程池结构*/typedef struct Pool{ pthread_mutex_t queue_lock;//互斥量 pthread_cond_t queue_ready;//条件变量 /*链表结构，线程池中所有等待任务*/ CThreadWorker *queue_head; int shutdown;//是否销毁线程池0为不销毁,1为销毁； pthread_t *threadid; int max_thread_num;//线程池中允许的活动线程数目； int cur_queue_size;//当前等待队列的任务数目；}CThreadPool;int pool_add_worker(void (*process)(void *arg),void *ptr);void *thread_routine(void *arg);static CThreadPool *pool =NULL;//线程池void pool_init(int max_thread_num){ //采用动态方式初始化条件变量和互斥量 pool = (CThreadPool*)malloc(sizeof(CThreadPool)); pthread_mutex_init(&(pool->queue_lock),NULL); pthread_cond_init(&(pool->queue_ready),NULL); pool->queue_head =NULL; pool->max_thread_num=max_thread_num; pool->cur_queue_size=0; pool->shutdown=0; pool->threadid = (pthread_t*)malloc(max_thread_num*sizeof(pthread_t)); int i = 0; for(i=0;i threadid[i]),NULL,thread_routine,NULL);//创建线程； } }/*int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);void *(*start_routine) (void *)回调函数，指向返回值为指针的函数*//*向线程池中加入任务*/int pool_add_worker(void (*process)(void *arg),void *ptr){ //构造一个新任务 CThreadWorker *newworker=(CThreadWorker*)malloc(sizeof(CThreadWorker)); newworker->process = process; newworker->arg = ptr; newworker->next =NULL; pthread_mutex_lock (&(pool->queue_lock)); /*将任务加入到等待队列中*/ CThreadWorker *member = pool->queue_head; if (member != NULL) { while (member->next != NULL) { member = member->next; } member->next = newworker; } else { pool->queue_head = newworker; } assert (pool->queue_head != NULL); pool->cur_queue_size++; pthread_mutex_unlock (&(pool->queue_lock)); /*好了，等待队列中有任务了，唤醒一个等待线程； 注意如果所有线程都在忙碌，这句没有任何作用*/ pthread_cond_signal (&(pool->queue_ready)); return 0; }/*销毁线程池，等待队列中的任务不会再被执行，但是正在运行的线程会一直 把任务运行完后再退出*/ int pool_destroy () { if (pool->shutdown) return -1;/*防止两次调用*/ pool->shutdown = 1; /*唤醒所有等待线程，线程池要销毁了*/ pthread_cond_broadcast (&(pool->queue_ready)); /*阻塞等待线程退出，否则就成僵尸了*/ int i; for (i = 0; i < pool->max_thread_num; i++) pthread_join (pool->threadid[i], NULL); free (pool->threadid); /*销毁等待队列*/ CThreadWorker *head = NULL; while (pool->queue_head != NULL) { head = pool->queue_head; pool->queue_head = pool->queue_head->next; free (head); } /*条件变量和互斥量也别忘了销毁*/ pthread_mutex_destroy(&(pool->queue_lock)); pthread_cond_destroy(&(pool->queue_ready)); free (pool); /*销毁后指针置空是个好习惯*/ pool=NULL; return 0; } /*pthread_t pthread_self(void);函数作用：获得线程自身的ID。pthread_t的类型为unsigned long int，所以在打印的时候要使用%lu方式，否则将产生奇怪的结果*/void * thread_routine (void *arg) { printf ("starting thread ---------------------%llu\n", pthread_self ()); while (1) { pthread_mutex_lock (&(pool->queue_lock)); /*如果等待队列为0并且不销毁线程池，则处于阻塞状态; 注意 pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁*/ while (pool->cur_queue_size == 0 && !pool->shutdown) { printf ("thread <%llu> is waiting\n", pthread_self ()); pthread_cond_wait (&(pool->queue_ready), &(pool->queue_lock)); } /*线程池要销毁了*/ if (pool->shutdown) { /*遇到break,continue,return等跳转语句，千万不要忘记先解锁*/ pthread_mutex_unlock (&(pool->queue_lock)); printf ("thread <%llu> will exit\n", pthread_self ()); pthread_exit (NULL); } printf ("thread <%llu> is starting to work\n", pthread_self ()); /*assert是调试的好帮手*/ assert (pool->cur_queue_size != 0); assert (pool->queue_head != NULL); /*等待队列长度减去1，并取出链表中的头元素*/ pool->cur_queue_size--; CThreadWorker *worker = pool->queue_head; pool->queue_head = worker->next; /*调用回调函数，执行任务*/ pthread_mutex_unlock (&(pool->queue_lock)); //这里为什么用这种表达方式；还有别的写法吗? (*(worker->process)) (worker->arg); free (worker); worker = NULL; } /*这一句应该是不可达的*/ pthread_exit (NULL); }void myprocess (void *arg) { printf ("threadid is <%llu>, working on task %llu\n", pthread_self (),*(int *) arg); sleep (1);/*休息一秒，延长任务的执行时间*/ // return NULL; } int main (int argc, char **argv) { pool_init (3);/*线程池中最多三个活动线程*/ // getchar(); sleep(50); /*连续向池中投入10个任务*/ int *workingnum = (int *) malloc (sizeof (int) * 10); int i; for (i = 0; i < 10; i++) { workingnum[i] = i; pool_add_worker (myprocess, &workingnum[i]); } /*等待所有任务完成*/ sleep (50); /*销毁线程池*/ pool_destroy (); free (workingnum); return 0; }#ifdef __cplusplus}#endif#endif