实现原理
程序启动之前,创建一定数量的线程,放入空闲的队列中,初始化线程池。
这些线程均处于阻塞状态,只占一点内存,不占用cpu。
当任务到达就从线程池中取出一个空闲线程,将任务传入此线程中运行。
当所有的线程都处在处理任务的时候,线程池将自动创建一定数量的新线程,用于处理更多的任务。
执行完任务的线程也并不退出,而是继续在线程池中等待下一次任务。
但大部分线程处于阻塞状态时,线程池将自动销毁一部分线程,回收系统资源。
组成部分
-
线程管理器
-
用于创建并管理线程池。
-
-
工作线程
-
线程池中实际执行任务的线程。在初始化线程时会预先创建好固定数目的线程在池中,这些初始化的线程一般处于空闲状态,一般不占用CPU,占用较小的内存空间。
-
-
任务接口
-
每个任务必须实现的接口,当线程池的任务队列中有可执行任务时,被空闲的工作线程调去执行,把任务抽象出来形成接口,可以做到线程池与具体的任务无关。
-
-
任务队列
-
用来存放没有处理的任务,提供一种缓冲机制
-
实现这种结构有好几种方法,常用的是队列,主要运用先进先出原理,另外一种是链表之类的数据结构,可以动态的为它分配内存空间,应用中比较灵活
-
简单实现
程序由三个文件组成,分别是thread_pool.h, thread_pool.c和test.c组成。
thread_pool.h如下:
#includestruct job { void * (*callback_function)(void *arg); void *arg; struct job *next; }; struct threadpool { int thread_num; int queue_max_num; struct job *head; struct job *tail; pthread_t *pthreads; pthread_mutex_t mutex; pthread_cond_t queue_empty; pthread_cond_t queue_not_empty; pthread_cond_t queue_not_full; int queue_cur_num; int queue_close; int pool_close; }; struct threadpool *threadpool_init(int thread_num, int queue_max_num); int threadpool_add_job(struct threadpool *pool, void *(*callback_function)(void *arg), void *arg); int threadpool_destroy(struct threadpool *pool); void *threadpool_function(void *arg); 123456789101112131415161718192021222324252627282930
thread_pool.c如下:
#include#include #include #include "thread_pool.h" struct threadpool *threadpool_init(int thread_num, int queue_max_num) { struct threadpool *pool = NULL; do { pool = (struct threadpool *)calloc(1, sizeof(struct threadpool)); if (!pool) { printf("calloc error: %m\n"); break; } pool->thread_num = thread_num; pool->queue_max_num = queue_max_num; pool->queue_cur_num = 0; pool->head = NULL; pool->tail = NULL; if (pthread_mutex_init(&(pool->mutex), NULL)) { printf("init mutex error: %m\n"); break; } if (pthread_cond_init(&(pool->queue_empty), NULL)) { printf("init queue_empty error: %m\n"); break; } if (pthread_cond_init(&(pool->queue_not_empty), NULL)) { printf("init queue_not_empty error: %m\n"); break; } if (pthread_cond_init(&(pool->queue_not_full), NULL)) { printf("init queue_not_full error: %m\n"); break; } pool->pthreads = calloc(1, sizeof(pthread_t) * thread_num); if (!pool->pthreads) { printf("calloc pthreads error: %m\n"); break; } pool->queue_close = 0; pool->pool_close = 0; int i; for (i = 0; i thread_num; i++) { pthread_create(&(pool->pthreads[i]), NULL, threadpool_function, (void *)pool); } return pool; } while (0); return NULL; } int threadpool_add_job(struct threadpool *pool, void *(*callback_function)(void *arg), void *arg) { assert(pool != NULL); assert(callback_function != NULL); assert(arg != NULL); pthread_mutex_lock(&(pool->mutex)); while ((pool->queue_cur_num == pool->queue_max_num) && !(pool->queue_close || pool->pool_close)) { pthread_cond_wait(&(pool->queue_not_full), &(pool->mutex)); } if (pool->queue_close || pool->pool_close) { pthread_mutex_unlock(&(pool->mutex)); return -1; } struct job *pjob = (struct job*) calloc(1, sizeof(struct job)); if (!pjob) { pthread_mutex_unlock(&(pool->mutex)); return -1; } pjob->callback_function = callback_function; pjob->arg = arg; pjob->next = NULL; if (pool->head == NULL) { pool->head = pool->tail = pjob; pthread_cond_broadcast(&(pool->queue_not_empty)); } else { pool->tail->next = pjob; pool->tail = pjob; } pool->queue_cur_num++; pthread_mutex_unlock(&(pool->mutex)); return 0; } void *threadpool_function(void *arg) { struct threadpool *pool = (struct threadpool *)arg; struct job *pjob = NULL; while (1) { pthread_mutex_lock(&(pool->mutex)); while ((pool->queue_cur_num == 0) && !pool->pool_close) { pthread_cond_wait(&(pool->queue_not_empty), &(pool->mutex)); } if (pool->pool_close) { pthread_mutex_unlock(&(pool->mutex)); pthread_exit(NULL); } pool->queue_cur_num--; pjob = pool->head; if (pool->queue_cur_num == 0) { pool->head = pool->tail = NULL; } else { pool->head = pjob->next; } if (pool->queue_cur_num == 0) { pthread_cond_signal(&(pool->queue_empty)); } if (pool->queue_cur_num == pool->queue_max_num - 1) { pthread_cond_broadcast(&(pool->queue_not_full)); } pthread_mutex_unlock(&(pool->mutex)); (*(pjob->callback_function))(pjob->arg); free(pjob); pjob = NULL; } } int threadpool_destroy(struct threadpool *pool) { assert(pool != NULL); pthread_mutex_lock(&(pool->mutex)); if (pool->queue_close || pool->pool_close) { pthread_mutex_unlock(&(pool->mutex)); return -1; } pool->queue_close = 1; while (pool->queue_cur_num != 0) { pthread_cond_wait(&(pool->queue_empty), &(pool->mutex)); } pool->pool_close = 1; pthread_mutex_unlock(&(pool->mutex)); pthread_cond_broadcast(&(pool->queue_not_empty)); pthread_cond_broadcast(&(pool->queue_not_full)); int i; for (i = 0; i thread_num; i++) { pthread_join(pool->pthreads[i], NULL); } pthread_mutex_destroy(&(pool->mutex)); pthread_cond_destroy(&(pool->queue_empty)); pthread_cond_destroy(&(pool->queue_not_empty)); pthread_cond_destroy(&(pool->queue_not_full)); free(pool->pthreads); struct job *p; while (pool->head != NULL) { p = pool->head; pool->head = p->next; free(p); } free(pool); return 0; } 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166
test.c用于测试,如下:
#include#include "thread_pool.h" void* work(void* arg) { char *p = (char*) arg; printf("threadpool callback fuction : %s.\n", p); sleep(1); } int main(void) { struct threadpool *pool = threadpool_init(10, 20); threadpool_add_job(pool, work, "1"); threadpool_add_job(pool, work, "2"); threadpool_add_job(pool, work, "3"); threadpool_add_job(pool, work, "4"); threadpool_add_job(pool, work, "5"); threadpool_add_job(pool, work, "6"); threadpool_add_job(pool, work, "7"); threadpool_add_job(pool, work, "8"); threadpool_add_job(pool, work, "9"); threadpool_add_job(pool, work, "10"); threadpool_add_job(pool, work, "11"); threadpool_add_job(pool, work, "12"); threadpool_add_job(pool, work, "13"); threadpool_add_job(pool, work, "14"); threadpool_add_job(pool, work, "15"); threadpool_add_job(pool, work, "16"); threadpool_add_job(pool, work, "17"); threadpool_add_job(pool, work, "18"); threadpool_add_job(pool, work, "19"); threadpool_add_job(pool, work, "20"); threadpool_add_job(pool, work, "21"); threadpool_add_job(pool, work, "22"); threadpool_add_job(pool, work, "23"); threadpool_add_job(pool, work, "24"); threadpool_add_job(pool, work, "25"); threadpool_add_job(pool, work, "26"); threadpool_add_job(pool, work, "27"); threadpool_add_job(pool, work, "28"); threadpool_add_job(pool, work, "29"); threadpool_add_job(pool, work, "30"); threadpool_add_job(pool, work, "31"); threadpool_add_job(pool, work, "32"); threadpool_add_job(pool, work, "33"); threadpool_add_job(pool, work, "34"); threadpool_add_job(pool, work, "35"); threadpool_add_job(pool, work, "36"); threadpool_add_job(pool, work, "37"); threadpool_add_job(pool, work, "38"); threadpool_add_job(pool, work, "39"); threadpool_add_job(pool, work, "40"); sleep(5); threadpool_destroy(pool); return 0; } 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
执行如下编译命令: gcc test.c thread_pool.c -lpthread
程序运行结果如下:
-> % ./a.out threadpool callback fuction : 1. threadpool callback fuction : 7. threadpool callback fuction : 8. threadpool callback fuction : 2. threadpool callback fuction : 3. threadpool callback fuction : 4. threadpool callback fuction : 5. threadpool callback fuction : 6. threadpool callback fuction : 9. threadpool callback fuction : 10. threadpool callback fuction : 12. threadpool callback fuction : 16. threadpool callback fuction : 11. threadpool callback fuction : 18. threadpool callback fuction : 14. threadpool callback fuction : 15. threadpool callback fuction : 17. threadpool callback fuction : 13. threadpool callback fuction : 19. threadpool callback fuction : 20. threadpool callback fuction : 21. threadpool callback fuction : 23. threadpool callback fuction : 24. threadpool callback fuction : 22. threadpool callback fuction : 26. threadpool callback fuction : 27. threadpool callback fuction : 28. threadpool callback fuction : 25. threadpool callback fuction : 29. threadpool callback fuction : 30. threadpool callback fuction : 31. threadpool callback fuction : 32. threadpool callback fuction : 33. threadpool callback fuction : 34. threadpool callback fuction : 35. threadpool callback fuction : 36. threadpool callback fuction : 37. threadpool callback fuction : 38. threadpool callback fuction : 39. threadpool callback fuction : 40.
以上就是
为各位朋友分享的 相关内容。想要了解更多Linux相关知识记得关注公众号“良许Linux”,或扫描下方二维码进行关注,更多 等着你!