Linux-C 编程 / 多线程 / 一个简洁可靠的线程池实现

发布时间：2022-10-12 12:58:10 所属栏目：Linux 来源：

导读：　　哈喽，我是老吴，我又来分享学习心得了。另外，为了更好地体现公众号的核心价值观，从本文开始，我会在文末新增分享一些非技术相关的内容，欢迎大家参与讨论。

　　一、简介 github.com/Pithikos/C-Thread-P

　　哈喽，我是老吴，我又来分享学习心得了。另外，为了更好地体现公众号的核心价值观，从本文开始，我会在文末新增分享一些非技术相关的内容，欢迎大家参与讨论。

　　一、简介 github.com/Pithikos/C-Thread-Pool

　　这是一个简单小巧的C语言线程池实现，在 Github 上有 1.1K 的 star，很适合用来学习 Linux 的多线程编程。

　　另外，里面还涉及到了信号、队列、同步等知识点，代码读起来还是挺过瘾的。

　　特点：二、使用快速上手

　　example.c：

　　#include "thpool.h"

　　void task(void *arg){
　　 printf("Thread #%u working on %d\n", (int)pthread_self(), (int) arg);
　　}
　　int main(){
　　
　　 puts("Making threadpool with 4 threads");
　　 threadpool thpool = thpool_init(4);
　　 puts("Adding 10 tasks to threadpool");
　　 int i;
　　 for (i=0; i<8; i++){
　　 thpool_add_work(thpool, task, (void*)(uintptr_t)i);
　　 };
　　 thpool_wait(thpool);
　　 puts("Killing threadpool");
　　 thpool_destroy(thpool);
　　
　　 return 0;
　　}
　　运行效果：

　　$ gcc example.c thpool.c -D THPOOL_DEBUG -pthread -o example
　　$ ./example
　　Making threadpool with 4 threads
　　THPOOL_DEBUG: Created thread 0 in pool
　　THPOOL_DEBUG: Created thread 1 in pool
　　THPOOL_DEBUG: Created thread 2 in pool
　　THPOOL_DEBUG: Created thread 3 in pool
　　Adding 10 tasks to threadpool
　　Thread #1509455616 working on 0
　　Thread #1509455616 working on 4
　　Thread #1509455616 working on 5
　　Thread #1492670208 working on 2
　　Thread #1492670208 working on 7
　　Thread #1509455616 working on 6
　　Thread #1501062912 working on 1
　　Thread #1517848320 working on 3
　　Killing threadpool
　　代码分析：

　　API 简介

　　线程池linux_net 线程池 (线程池)threadpool_线程池 linux

　　三、内部实现整体把握

　　核心代码就是 2 个文件： thpool.c 和 thpool.h。

　　分解 thpool.c

　　7 个公共函数：

　　struct thpool_* thpool_init(int num_threads)
　　int thpool_add_work(thpool_* thpool_p, void (*function_p)(void*), void* arg_p)
　　void thpool_wait(thpool_* thpool_p)
　　void thpool_destroy(thpool_* thpool_p)
　　void thpool_pause(thpool_* thpool_p)
　　void thpool_resume(thpool_* thpool_p)
　　int thpool_num_threads_working(thpool_* thpool_p)
　　正好就是前面说过的 7 个 API，稍后重点分析。

　　5 个自定义的数据结构：

　　// 描述一个信号量
　　typedef struct bsem {...} bsem;
　　// 描述一个任务
　　typedef struct job {...} job;
　　// 描述一个任务队列
　　typedef struct jobqueue {...} jobqueue;
　　// 描述一个线程
　　typedef struct thread {...} thread;
　　// 描述一个线程池
　　typedef struct thpool_ {...} thpool_;
　　14 个私有函数：

　　// 构造 struct thread，并调用 pthread_create() 创建线程
　　static int thread_init (thpool_* thpool_p, struct thread** thread_p, int id)
　　// 当线程被暂停时会在这里休眠
　　static void thread_hold(int sig_id)
　　// 线程在此函数中执行任务
　　static void* thread_do(struct thread* thread_p)
　　// 销毁 struct thread
　　static void thread_destroy (thread* thread_p)
　　// 任务队列相关的操作集合
　　static int jobqueue_init(jobqueue* jobqueue_p)
　　static void jobqueue_clear(jobqueue* jobqueue_p)
　　static void jobqueue_push(jobqueue* jobqueue_p, struct job* newjob)
　　static struct job* jobqueue_pull(jobqueue* jobqueue_p)
　　static void jobqueue_destroy(jobqueue* jobqueue_p)
　　// 信号量相关的操作集合
　　static void bsem_init(bsem *bsem_p, int value)
　　static void bsem_reset(bsem *bsem_p)
　　static void bsem_post(bsem *bsem_p)
　　static void bsem_post_all(bsem *bsem_p)
　　static void bsem_wait(bsem* bsem_p)
　　核心 API 的实现1. thpool_init()

　　该函数用于创建一个线程池，先明确线程池的定义：

　　typedef struct thpool_{
　　 thread**   threads;                  /* pointer to threads        */
　　 volatile int num_threads_alive;      /* threads currently alive   */
　　 volatile int num_threads_working;    /* threads currently working */
　　 pthread_mutex_t thcount_lock;       /* used for thread count etc */
　　 pthread_cond_t threads_all_idle;    /* signal to thpool_wait     */
　　 jobqueue jobqueue;                  /* job queue                 */
　　} thpool_;
　　thpool_init() 的实现思路：

　　分配 struct thpool_：

　　2. 初始化 struct thpool_：

　　创建用户指定数目的线程，用一个二级指针来指向这一组线程： thread_init() 会调用 pthread_create() 创建线程，执行函数为 thread_do()

　　3. 返回 struct thpool_ *;

　　thread_do() 的实现思路：

　　注册信号 SIGUSR1 的处理函数，用于后续实现线程池的 pause 功能;

　　2. 进入循环 while(threads_keepalive)：

　　3. 如果已经没有线程在工作了，则广播通知 main 线程;

　　2. thpool_add_work()

　　该函数用于往线程池里添加一个任务，先明确任务的定义：

　　typedef struct job{
　　 struct job* prev; /* pointer to previous job */
　　 void   (*function)(void* arg); /* function pointer */
　　 void* arg; /* function's argument */
　　} job;
　　程序里是用队列来管理任务的，这里的 job 首先是一个队列节点，携带的数据是 function + arg。

　　thpool_add_work 的实现思路：

　　分配 struct job：

　　2. 初始化 struct job;

　　3. 添加到队列中：

　　3. thpool_pause() 和 thpool_resume()

　　thpool_pause() 用于暂停所有的线程，通过信号机制来实现：

　　void thpool_pause(thpool_* thpool_p) {
　　 int n;
　　 for (n=0; n < thpool_p->num_threads_alive; n++){
　　 pthread_kill(thpool_p->threads[n]->pthread, SIGUSR1);
　　 }
　　}
　　给所有工作线程发送 SIGUSR1，该信号的处理行为就是让线程休眠：

　　static void thread_hold(int sig_id) {
　　    (void)sig_id;
　　 threads_on_hold = 1;
　　 while (threads_on_hold){
　　 sleep(1);
　　 }
　　}
　　之需要 thpool_resume() 中，将 threads_on_hold = 0线程池linux，就可以让线程返回到原来被中止时的工作状态。

　　4. thpool_wait()

　　wait 的实现比较简单，只要还有任务或者还有线程处于工作状态，就执行 pthread 的 wait 操作：

　　while (thpool_p->jobqueue.len || thpool_p->num_threads_working) {
　　 pthread_cond_wait(&thpool_p->threads_all_idle, &thpool_p->thcount_lock);
　　 }
　　到此，我感觉已经没有太多难点了，感兴趣的小伙伴们可以自行查阅源码。

　　四、测试用例

　　优秀的开源项目通常会附带丰富的测试用例，此项目也不例外：

　　思考技术，也思考人生

　　要学习技术，更要学习如何生活。

　　最近在看的书：

　　《精要主义》

　　收获了什么？

　　身体是一种资产：

　　你和我各有一个苹果，如果我们交换苹果的话，我们还是只有一个苹果。但当你和我各有一个想法，我们交换想法的话，我们就都有两个想法了。

（编辑：云计算网_汕头站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!

鲜为人知但功能强悍的	怎样在 Ubuntu 22.04
Linux 上文件编辑器 E	嵌入式Linux文件系统详