良许Linux教程网 干货合集 Linux系统C线程池简单案例

Linux系统C线程池简单案例

作者: 良许 1,649阅读 0评论

使用并行/并发编程开发服务器是经常遇到得事情，下面良许教程网为大家分享一下Linux系统中 c 线程池的具体使用方法。

Linux系统C线程池简单案例

实现原理

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

这些线程均处于阻塞状态，只占一点内存，不占用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系统相关内容。想要了解更多Linux相关知识记得关注公众号“良许Linux”，或扫描下方二维码进行关注，更多干货等着你！

137e00002230ad9f26e78-265x300

本文由 良许Linux教程网 发布，可自由转载、引用，但需署名作者且注明文章出处。如转载至微信公众号，请在文末添加作者公众号二维码。
打赏 9赞
标签:
良许

作者: 良许

良许，世界500强企业Linux开发工程师，公众号【良许Linux】的作者，全网拥有超30W粉丝。个人标签：创业者，CSDN学院讲师，副业达人，流量玩家，摄影爱好者。
上一篇
下一篇

相关推荐

发表评论

联系我们

联系我们

公众号：良许Linux

在线咨询: QQ交谈

邮箱: yychuyu@163.com

关注微信
微信扫一扫关注我们

微信扫一扫关注我们

关注微博
返回顶部