• 基于C语言 -- 线程池实现

    项目地址

    GitHub - CHENLitterWhite/CPPWheel: CPP自封装的库

    专栏历史好文

    CMakeList.txt -- 编写保姆式教程_干饭小白的博客-CSDN博客

    C++实战-基于c++11新特性的mysql连接池_c++ mysql连接池_干饭小白的博客-CSDN博客

    C++项目实战--线程池代码讲解_std::shared_ptr_干饭小白的博客-CSDN博客

    C++项目实战-实际应用_干饭小白的博客-CSDN博客

    文章目录

    目录

    前言

    使用步骤

    前言

    线程池 -- 纯C版

    使用步骤

    1. //------------------------------------------//
    2. // 作者: 干饭小白
    3. // 时间: 2023-09-04
    4. //------------------------------------------//
    5. #pragma once
    6.  
    7. #ifdef __cplusplus
    8. extern "C"
    9. {
    10. #endif __cplusplus
    11.  
    12. #define DEFAULT_TIME       5
    13. #define MIN_WAIT_TASK_NUM  5
    14. #define DEFAULT_THREAD_NUM 2
    15.  
    16. #include 
    17. #include 
    18. #include 
    19. #include 
    20. #include 
    21. #include 
    22.  
    23. // 任务函数结构
    24. typedef struct {
    25.     void *(*function)(void*);
    26.     void *arg;
    27. }Task;
    28.  
    29. // 线程池管理者
    30. struct Threadpool {
    31.  
    32.     /* 开关 */
    33.     bool shutdown;
    34.  
    35.     /* 任务队列 */
    36.     int queue_front;                  // 队头
    37.     int queue_rear;                   // 队尾
    38.     int queue_size;                   // 队列大小
    39.     int queue_max_size;               // 最大的任务数
    40.     Task *task_queue;                 // 任务队列
    41.  
    42.     /* 工作线程状态 */
    43.     int min_thr_num;                  // 最小的线程数
    44.     int max_thr_num;                  // 最大线线程数
    45.     int live_thr_num;                 // 当前的活跃线程数
    46.     int busy_thr_num;                 // 当前忙碌的线程数
    47.     int wait_exit_thr_num;            // 当前正在等待退出的线程数
    48.     pthread_t *threads;               // 线程集合
    49.  
    50.     /* 线程集合的管理 */
    51.     pthread_t admin_tid;              // 线程管理者 -- 保持线程池中线程的相对平衡
    52.     pthread_mutex_t lock;             // 互斥锁
    53.     pthread_mutex_t thread_counter;   // 保证线程间竞争关系互斥 -- 正在忙碌的线程变量控制
    54.     pthread_cond_t queue_not_full;    // 唤醒任务可入队线程 -- 队列不为满
    55.     pthread_cond_t queue_not_empty;   // 唤醒工作线程取任务 -- 队列不为空
    56. };
    57.  
    58. /*-----------------------------------------------
    59. 函数名:     
    60.     threadpool_creat
    61. 函数说明:   
    62.     创建并初始化一个线程池
    63. 输入参数:   
    64.     min_thr_num(int)     最小的线程数量 
    65.     max_thr_num(int)     最大的线程数量 
    66.     queue_max_size(int)  任务队列大小
    67. 输出参数:
    68.     无
    69. 返回值:
    70.     返回线程池句柄,失败返回 NULL
    71. -----------------------------------------------*/
    72. Threadpool * threadpool_creat(int min_thr_num, int max_thr_num, int queue_max_size);
    73.  
    74. /*-----------------------------------------------
    75. 函数名:     
    76.     threadpool_thread
    77. 函数说明:   
    78.     任务处理线程函数
    79. 输入参数:   
    80.     threadpool(Threadpool) 线程池句柄
    81. 输出参数:
    82.     无
    83. 返回值:
    84.     无
    85. -----------------------------------------------*/
    86. void * threadpool_thread(void *threadpool);
    87.  
    88. /*-----------------------------------------------
    89. 函数名:     
    90.     threadpool_thread
    91. 函数说明:   
    92.     任务处理线程函数
    93. 输入参数:   
    94.     threadpool(Threadpool*) 线程池句柄
    95. 输出参数:
    96.     无
    97. 返回值:
    98.     无
    99. -----------------------------------------------*/
    100. void * admin_thread(void *threadpool);
    101.  
    102. /*-----------------------------------------------
    103. 函数名:     
    104.     threadpool_free
    105. 函数说明:   
    106.     释放资源
    107. 输入参数:   
    108.     pool(Threadpool*) 线程池句柄
    109. 输出参数:
    110.     无
    111. 返回值:
    112.     0   正常
    113.     -1  pool == NULL
    114. -----------------------------------------------*/
    115. int threadpool_free(Threadpool * pool);
    116.  
    117. /*-----------------------------------------------
    118. 函数名:     
    119.     threadpool_destory
    120. 函数说明:   
    121.     销毁线程池
    122. 输入参数:   
    123.     pool(Threadpool*) 线程池句柄
    124. 输出参数:
    125.     无
    126. 返回值:
    127.     0   正常
    128.     -1  pool == NULL
    129. -----------------------------------------------*/
    130. int threadpool_destory(Threadpool * pool);
    131.  
    132. /*-----------------------------------------------
    133. 函数名:     
    134.     threadpool_add_task
    135. 函数说明:   
    136.     向任务队列中添加事件
    137. 输入参数:   
    138.     pool(Threadpool*) 线程池句柄
    139.     void *(*function)(void *arg) 任务函数
    140.     arg(void *) 任务函数携带的参数
    141. 输出参数:
    142.     无
    143. 返回值:
    144.     无
    145. -----------------------------------------------*/
    146. int threadpool_add_task(Threadpool * pool, void *(*function)(void *arg), void *arg);
    147.  
    148. /*-----------------------------------------------
    149. 函数名:     
    150.     is_thread_alive
    151. 函数说明:   
    152.     判断一个线程是否存活
    153. 输入参数:   
    154.     tid(pthread_t) 线程pid
    155. 输出参数:
    156.     无
    157. 返回值:
    158.     存活 true
    159.     消亡 false
    160. -----------------------------------------------*/
    161. bool is_thread_alive(pthread_t tid);
    162.  
    163. #ifdef __cplusplus
    164. }
    165. #endif
    1. #include "ThreadPool.h"
    2.  
    3. Threadpool * threadpool_creat(int min_thr_num, int max_thr_num, int queue_max_size)
    4. {
    5.     Threadpool *pool = NULL;
    6.     
    7.     /* 使用do{}while(0) --> 实现goto机制,错误时即使跳出,并统一处理 */
    8.     do
    9.     {
    10.         /** 开启线程池空间 **/
    11.         pool = (Threadpool *)malloc(sizeof(Threadpool));
    12.         if(NULL == pool)
    13.         {
    14.             break;
    15.         }
    16.  
    17.         /** 初始化信息 **/
    18.         pool->min_thr_num = min_thr_num;
    19.         pool->max_thr_num = max_thr_num;
    20.         pool->live_thr_num = min_thr_num;
    21.         pool->busy_thr_num = 0;
    22.         pool->wait_exit_thr_num = 0;
    23.         pool->queue_front = 0;
    24.         pool->queue_rear = 0;
    25.         pool->queue_size = 0;
    26.         pool->queue_max_size = queue_max_size;
    27.         pool->shutdown = false;
    28.  
    29.         /** 分配工作线程空间 **/
    30.         pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * max_thr_num);
    31.         if(NULL == pool->threads)
    32.         {
    33.             break;
    34.         }
    35.         memset(pool->threads, 0, sizeof(pthread_t) * max_thr_num);
    36.  
    37.         /** 队列空间 **/
    38.         pool->task_queue = (Task *)malloc(sizeof(Task) * queue_max_size);
    39.         if(NULL == pool->task_queue)
    40.         {
    41.             break;
    42.         }
    43.  
    44.         /** 初始化互斥锁和条件变量 **/
    45.         if(pthread_mutex_init(&(pool->lock), NULL) != 0 ||
    46.            pthread_mutex_init(&(pool->thread_counter), NULL) != 0 ||
    47.            pthread_cond_init(&(pool->queue_not_empty), NULL) ||
    48.            pthread_cond_init(&(pool->queue_not_full), NULL) != 0)
    49.         {
    50.  
    51.             break;
    52.         } 
    53.  
    54.         /** 启动 min_thr_num 个工作线程 **/
    55.         for(int i = 0; i < min_thr_num; ++i)
    56.         {
    57.             pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
    58.         }
    59.  
    60.         /** 启动管理者线程 **/
    61.         pthread_create(&(pool->admin_tid), NULL, admin_thread, (void*)pool);
    62.  
    63.         return pool;
    64.  
    65.     }while(0);
    66.  
    67.     /* 释放pool的空间 */
    68.     threadpool_free(pool);
    69.  
    70.     return NULL;
    71. }
    72.  
    73. int threadpool_destory(Threadpool * pool)
    74. {
    75.     /* 容错性判断 */
    76.     if(NULL == pool)
    77.     {
    78.         return -1;
    79.     }
    80.  
    81.     /* 线程标志位:true */
    82.     pool->shutdown = true;
    83.  
    84.     /* 销毁管理线程 */
    85.     pthread_join(pool->admin_tid, NULL);
    86.  
    87.     /* 通知存活的线程结束自己 */
    88.     for(int i = 0; i < pool->live_thr_num; ++i)
    89.     {
    90.         /** 唤醒所有被阻塞的线程 **/
    91.         pthread_cond_broadcast(&(pool->queue_not_empty));
    92.     }
    93.  
    94.     /* 等待线程结束,进行回收 */
    95.     for(int i = 0; i < pool->live_thr_num; ++i)
    96.     {
    97.         pthread_join(pool->threads[i], NULL);
    98.     }
    99.  
    100.     /* 释放资源 */
    101.     threadpool_free(pool);
    102.  
    103.     return 0;
    104. }
    105.  
    106. int threadpool_free(Threadpool * pool)
    107. {
    108.     /* 容错性判断 */
    109.     if(NULL == pool)
    110.     {
    111.         return -1;
    112.     }
    113.  
    114.     /* 销毁任务队列 */
    115.     if(pool->task_queue)
    116.     {
    117.         free(pool->task_queue);
    118.         pool->task_queue = NULL;
    119.     }
    120.  
    121.     /* 销毁工作线程集合 */
    122.     if(pool->threads)
    123.     {
    124.         free(pool->threads);
    125.         pool->threads = NULL;
    126.     }
    127.  
    128.     /* 销毁池管理空间 */
    129.     free(pool);
    130.     pool = NULL;
    131.  
    132.     return 0;
    133. }
    134.  
    135. void * admin_thread(void *threadpool)
    136. {
    137.     /* 容错处理 */
    138.     if(NULL == threadpool)
    139.     {
    140.         return NULL;
    141.     }
    142.  
    143.     /* 维护线程池的平衡 */
    144.     Threadpool *pool = (Threadpool *)threadpool;
    145.     while(!pool->shutdown)
    146.     {
    147.  
    148.         /** 每隔DEFAULT_TIME进行一次 **/
    149.         sleep(DEFAULT_TIME);
    150.  
    151.         /** 获取当前状态下活跃数和任务数 -- 多线程访问同一个变量(互斥锁) **/
    152.         pthread_mutex_lock(&(pool->lock));
    153.         int queue_size = pool->queue_size;
    154.         int live_thr_num = pool->live_thr_num;
    155.         pthread_mutex_unlock(&(pool->lock));
    156.  
    157.         /** 获取忙碌数 **/
    158.         pthread_mutex_lock(&(pool->thread_counter));
    159.         int busy_thr_num = pool->busy_thr_num;
    160.         pthread_mutex_unlock(&(pool->thread_counter));
    161.  
    162.         /** 创建新线程:正在等待的任务 >= 最小允许等待的任务数 && 存活线程数 < 最大线程数 **/
    163.         if(queue_size >= MIN_WAIT_TASK_NUM && live_thr_num < pool->max_thr_num)
    164.         {
    165.             pthread_mutex_lock(&(pool->lock));
    166.             int add = 0;
    167.             for(int i = 0; i < pool->max_thr_num && add < DEFAULT_THREAD_NUM && pool->live_thr_num < pool->max_thr_num; ++i)
    168.             {
    169.                 if(pool->threads[i] == 0 || !is_thread_alive(pool->threads[i]))
    170.                 {
    171.                     pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void*)pool);
    172.                     add++;
    173.                     pool->live_thr_num++;
    174.                 }
    175.             }
    176.             pthread_mutex_unlock(&(pool->lock));
    177.         }
    178.  
    179.         /** 销毁空闲的线程: 忙碌数*2 < 存活数(一半以上的空闲) && 存活数 > 最小线程数 **/
    180.         if((busy_thr_num * 2) < live_thr_num && live_thr_num > pool->min_thr_num)
    181.         {
    182.             pthread_mutex_lock(&(pool->lock));
    183.             pool->wait_exit_thr_num = DEFAULT_THREAD_NUM;
    184.             pthread_mutex_unlock(&(pool->lock));
    185.         }
    186.  
    187.         for(int i = 0; i < DEFAULT_THREAD_NUM; ++i)
    188.         {
    189.             /*** 通知处于空闲的线程自杀 ***/
    190.             pthread_cond_signal(&(pool->queue_not_empty));
    191.         }
    192.  
    193.     }
    194.  
    195.     return NULL;
    196. }
    197.  
    198. bool is_thread_alive(pthread_t tid)
    199. {
    200.     int kill_rc = pthread_kill(tid, 0);
    201.     if(ESRCH == kill_rc)
    202.     {
    203.         return false;
    204.     }
    205.  
    206.     return true;
    207. }
    208.  
    209. void * threadpool_thread(void *threadpool)
    210. {
    211.     /* 容错处理 */
    212.     if(NULL == threadpool)
    213.     {
    214.         return NULL;
    215.     }
    216.  
    217.     Threadpool *pool = (Threadpool *)threadpool;
    218.     Task task;
    219.  
    220.     while(true)
    221.     {
    222.         pthread_mutex_lock(&(pool->lock));
    223.  
    224.         /* 无任务:阻塞等待 */
    225.         while((pool->queue_size == 0 && !pool->shutdown))
    226.         {
    227.             /** 阻塞等待 **/
    228.             pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
    229.  
    230.             /** 是否是自杀任务 **/
    231.             if(pool->wait_exit_thr_num > 0)
    232.             {
    233.                 pool->wait_exit_thr_num--;
    234.  
    235.                 /** 判断线程池中的线程数量是否大于最小线程数:是则结束当前线程 **/
    236.                 if(pool->live_thr_num > pool->min_thr_num)
    237.                 {
    238.                     pool->live_thr_num--;
    239.                     pthread_mutex_unlock(&(pool->lock));
    240.                     pthread_exit(NULL);
    241.                 }
    242.             }
    243.         }
    244.  
    245.         /* 是否销毁池 */
    246.         if(pool->shutdown)
    247.         {
    248.             pthread_mutex_unlock(&(pool->lock));
    249.             pthread_exit(NULL);
    250.         }
    251.  
    252.         /* 有任务:执行任务 */
    253.         task.function = pool->task_queue[pool->queue_front].function;
    254.         task.arg = pool->task_queue[pool->queue_front].arg;
    255.  
    256.         /* 维护循环队列平衡 */
    257.         pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size;
    258.         pool->queue_size--; 
    259.         
    260.         /* 通知队列不未满 */
    261.         pthread_cond_broadcast(&(pool->queue_not_full));
    262.         
    263.         pthread_mutex_unlock(&(pool->lock));
    264.  
    265.         /* 执行开始:忙碌线程+1 */
    266.         pthread_mutex_lock(&(pool->thread_counter));
    267.         pool->busy_thr_num++;
    268.         pthread_mutex_unlock(&(pool->thread_counter));
    269.         
    270.         /* 执行任务 */
    271.         (*(task.function))(task.arg);
    272.  
    273.         /* 执行结束:忙碌线程-1 */
    274.         pthread_mutex_lock(&(pool->thread_counter));
    275.         pool->busy_thr_num--;
    276.         pthread_mutex_unlock(&(pool->thread_counter));
    277.     }
    278.  
    279.     pthread_exit(NULL);
    280. }
    281.  
    282. int threadpool_add_task(Threadpool * pool, void *(*function)(void *arg), void *arg)
    283. {
    284.     if(NULL == pool)
    285.     {
    286.         return -1;
    287.     }
    288.     
    289.     pthread_mutex_lock(&(pool->lock));
    290.  
    291.     /* 任务队列满时需要等到 */
    292.     while((pool->queue_size == pool->queue_max_size) && (!pool->shutdown))
    293.     {
    294.         pthread_cond_wait(&(pool->queue_not_full), &(pool->lock));
    295.     }
    296.  
    297.     /* 线程池处于关闭状态 */
    298.     if(pool->shutdown)
    299.     {
    300.         pthread_mutex_unlock(&(pool->lock));
    301.     }
    302.  
    303.     /* 清空残留 */
    304.     if(pool->task_queue[pool->queue_rear].arg != NULL)
    305.     {
    306.         free(pool->task_queue[pool->queue_rear].arg);
    307.         pool->task_queue[pool->queue_rear].arg = NULL;
    308.     }
    309.  
    310.     /* 添加任务 */
    311.     pthread_cond_signal(&(pool->queue_not_empty));
    312.     pthread_mutex_unlock(&(pool->lock));
    313.  
    314.     return 0;
    315. }

  • 相关阅读:
    Programming Languages PartC Week1学习笔记——Ruby与面向对象编程
    包埋紫杉醇的Pluronic P85/聚乳酸(PLA-P85-PLA)纳米粒子|制备方法
    LabVIEW专栏五、网口
    hexo 使用hexo g -d报错
    shiro详解-shiro史上最全学习笔记
    数据结构——排序算法——桶排序
    STS中打开Ibatis的xml文件提示错误
    VE对环境和社会的贡献
    QT单元测试初探
    golang select 机制
  • 原文地址：https://blog.csdn.net/weixin_46120107/article/details/132948426