线程池
有时我们会需要大量线程来处理一些相互独立的任务，为了避免频繁的申请释放线程所带来的开销，我们可以使用线程池。下面是一个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_tqueue_ready; 20: }tpool_t;21:22: /*23: * @brief 创建线程池 24: * @param max_thr_num 最大线程数25: * @return 0: 成功 其他: 失败26: */27: int28: tpool_create(int max_thr_num);29:30: /*31: * @brief 销毁线程池 32: */33: void34: tpool_destroy();35:36: /*37: * @brief 向线程池中添加任务38: * @param routine 任务函数指针39: * @param arg 任务函数参数40: * @return 0: 成功 其他:失败 41: */42: int43: 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: int42: 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, 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: void87: 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, 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: int13: 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之后，仅将当前正在执行的任务完成之后就会退出，我们也可以修改代码使得线程池在执行完任务链表中所有任务后再退出。
好凌乱的排版，88 排版确实有点乱了 不过谢谢楼主分享的
页:
[1]