Linux系统线程池的示例分析

Linux系统线程池的示例分析，针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。

实现原理

程序启动之前，创建一定数量的线程，放入空闲的队列中，初始化线程池。

这些线程均处于阻塞状态，只占一点内存，不占用cpu。

当任务到达就从线程池中取出一个空闲线程，将任务传入此线程中运行。

当所有的线程都处在处理任务的时候，线程池将自动创建一定数量的新线程，用于处理更多的任务。

执行完任务的线程也并不退出，而是继续在线程池中等待下一次任务。

但大部分线程处于阻塞状态时，线程池将自动销毁一部分线程，回收系统资源。

组成部分

• 线程管理器

• 用于创建并管理线程池。

• 工作线程

• 线程池中实际执行任务的线程。在初始化线程时会预先创建好固定数目的线程在池中，这些初始化的线程一般处于空闲状态，一般不占用CPU，占用较小的内存空间。

• 任务接口

• 每个任务必须实现的接口，当线程池的任务队列中有可执行任务时，被空闲的工作线程调去执行，把任务抽象出来形成接口，可以做到线程池与具体的任务无关。

• 任务队列

• 用来存放没有处理的任务，提供一种缓冲机制

• 实现这种结构有好几种方法，常用的是队列，主要运用先进先出原理，另外一种是链表之类的数据结构，可以动态的为它分配内存空间，应用中比较灵活

简单实现

程序由三个文件组成，分别是thread_pool.h, thread_pool.c和test.c组成。

thread_pool.h如下：

 #include  struct 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系统线程池的示例分析问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注行业资讯频道了解更多相关知识。