基于c语言实现一个线程池以及线程池基本原理

在POSIX API的标准下一个线程需要8M的内存空间1024M的内存空间可以开辟128个线程。如果是16G的内存可以开16×128 2048个线程。显然一次性开辟大量的线程对我们的计算机资源的消耗是十分大的。为了避免这种消耗我们可以使用线程池线程池可以提前开辟一些线程后续如果又任务产生则可以通过调度线程池中的空闲的线程来直接执行对应的任务。1 线程池组成1.1 任务队列任务队列的包括了任务数据以及对应回调函数该回调函数用于对任务数据进行消费根据需求来处理该任务数据单个任务的结构体定义如下:typedefstructTask{void(*task_handler)(void*data);//回调函数用于处理任务数据void*task_data;//任务数据structTask*next;//指向下一个任务}Task;1.2 工作队列工作队列中的每个节点就是我们创建的线程每个线程都需要从任务队列中提取任务然后通过每个任务的回调函数来处理这个任务。每个工作队列的的结构体定义如下typedefstructWorker{pthread_tthread_id;//工作的线程idstructManager*manager;//指向管理者structWorker*next;//下一个工作线程structWorker*prev;//上一个工作线程}Worker;1.3 管理者管理者是线程池的调度中枢它不直接处理任务而是通过锁 条件变量实现对任务队列和工作线程的线程安全协调。其结构体定义如下typedefstructManager{structTaskQueue*task_queue;// 任务队列FIFOstructWorkerQueue*worker_queue;// 工作线程队列pthread_mutex_tmutex;// 互斥锁保护队列和状态pthread_cond_tcond;// 条件变量用于阻塞/唤醒工作线程intshutdown;// 销毁标志1 表示线程池即将关闭}Manager;管理者是调度任务队列和工作队列的组件是让线程池可以正常运转的核心通过管理者我们可以知道哪些工作线程是空闲的任务来了就通知工作线程来处理。还有哪些任务正在被工作线程处理正在处理的任务就不用通知别的工作线程处理了。在整个调度过程中任务队列和工作队列不直接沟通需要通过管理者来协调。比如当你去高档饭店吃饭后厨有厨师长和许多的厨师厨师长可以理解为这里的管理者厨师就是工作线程当菜单栏任务队列没有任务的时候所有的厨师可以坐着休息当来了菜单厨师长管理者就协调一个人来做菜。结合上述定义的结构体锁mutex用于确保做菜的厨师们工作线程不会乱比如一个厨师工作线程做了西红柿炒蛋这道菜并且菜单栏任务队列上没有这道菜了其他厨师工作线程就不会做了。对于条件变量cond可以类似为一个指示灯如果厨师长管理者发现新的菜单入队就唤醒条件变量cond打开指示灯通知一个厨师工作线程来拿到这个任务没有任务则关闭指示灯所有的厨师可以直接休息。除此之外厨师长知道饭店打烊了就通过指示灯来通知所有的厨师不用准备了直接下班。示意图如下当菜单没有菜名的时候所有的厨师可以坐着休息下不用在灶台前等候但菜单栏有菜名的时候厨师长通过条件变量来告诉厨师取任务厨师取到自己需要做的菜名但是每个菜都有自己的做法厨师需要根据菜名来做对应我们的任务队列中的每个任务都有处理任务数据的回调函数通过这个回调函数来处理任务数据。2 接口实现2.1 线程池的初始化线程池的初始化参数包含了管理者以及需要开辟的工作线程个数主要是对管理者的相关参数进行设置以及通过pthread_create()来创建每个工作队列节点的线程。/** * brief 创建并初始化线程池 * param manager 管理者结构体指针调用者分配内存 * param worker_count 要创建的工作线程数量 * return 0 成功-1 参数错误-2 工作线程结构体分配失败-3 线程创建失败 */intthreadpool_create(Manager*manager,size_tworker_count){// 1. 参数校验管理者指针不能为空线程数量至少为1if(!manager||worker_count1)return-1;// 2. 初始化基本状态关闭标志清0表示线程池正在运行manager-shutdown0;// 3. 为任务队列和工作队列分配内存manager-task_queuemalloc(sizeof(TaskQueue));// 任务队列结构体manager-worker_queuemalloc(sizeof(WorkerQueue));// 工作线程队列结构体// 4. 队列初始化为空头尾指针置空manager-task_queue-headNULL;manager-task_queue-tailNULL;manager-worker_queue-headNULL;// 5. 初始化互斥锁和条件变量用于线程安全调度pthread_cond_init(manager-cond,NULL);// 条件变量默认属性pthread_mutex_init(manager-mutex,NULL);// 互斥锁默认属性// 6. 循环创建指定数量的工作线程for(size_ti0;iworker_count;i){// 6.1 为单个工作线程结构体动态分配内存Worker*sin_workermalloc(sizeof(Worker));if(!sin_worker){perror(malloc);// 打印分配失败原因free(sin_worker);// 释放可能部分分配的内存此处为空return-2;// 返回分配失败错误码}// 6.2 让工作线程知道自己属于哪个管理者sin_worker-managermanager;// 6.3 创建线程入口函数为 worker_threadif(pthread_create(sin_worker-thread_id,NULL,worker_thread,sin_worker)){perror(pthread create);// 打印线程创建失败原因free(sin_worker);// 释放已分配的工作线程节点return-3;// 返回线程创建失败错误码}// 6.4 将新创建的工作线程节点插入到工作队列头部链表头插法// LIST_INSERT 是一个自定义宏功能为将第一个参数节点插入第二个参数链表头部LIST_INSERT(sin_worker,manager-worker_queue-head);}return0;}2.2 工作线程的实现每个工作队列节点都对应一个后台的线程用于从任务队列拿出任务数据然后进行处理。/** * brief 工作线程入口函数每个工作线程启动时执行 * param arg 指向当前工作线程对应的 Worker 结构体 * return NULL */staticvoid*worker_thread(void*arg){// 1. 获取当前线程的管理信息包含任务队列、锁、条件变量等Worker*sin_worker(Worker*)arg;while(1)// 工作线程主循环持续等待和处理任务{// 2. 加锁所有对任务队列和关闭标志的操作必须在互斥锁保护下进行pthread_mutex_lock(sin_worker-manager-mutex);// 3. 等待任务到来或线程池关闭信号// 使用 while 而非 if 是为了防止虚假唤醒spurious wakeupwhile(!sin_worker-manager-task_queue-head!sin_worker-manager-shutdown){// 3.1 条件变量阻塞释放 mutex 并进入等待直到被 signal 或 broadcast 唤醒pthread_cond_wait(sin_worker-manager-cond,sin_worker-manager-mutex);// 被唤醒后会重新获得锁再次检查条件是否满足}// 4. 取出任务队列的第一个任务FIFOTask*sin_tasksin_worker-manager-task_queue-head;// 5. 处理关闭场景如果任务队列为空且关闭标志被设置表示线程池即将销毁if(!sin_tasksin_worker-manager-shutdown){pthread_mutex_unlock(sin_worker-manager-mutex);// 释放锁再退出break;// 跳出循环线程正常结束}// 6. 从队列中移除该任务更新头指针若队列变空则尾指针也置空sin_worker-manager-task_queue-headsin_task-next;if(!sin_worker-manager-task_queue-head)sin_worker-manager-task_queue-tailNULL;// 7. 解锁对队列的修改已完成释放锁让其他线程可以操作队列pthread_mutex_unlock(sin_worker-manager-mutex);// 8. 执行任务调用该任务预先设置的回调函数传入任务数据sin_task-task_handler(sin_task-task_data);// 注意任务结构体sin_task通常由主线程分配这里执行完后应由主线程或其他机制释放// 或者在回调函数内自行释放。此示例中未在这里释放可根据具体设计在回调中释放。}returnNULL;}2.3 任务队列中处理任务数据的回调函数将任务队列中的任务数据的回调函数写好这样当我工作线程拿到了任务直接调用对应的回调函数就可以完成任务数据的处理。这里以打印接收到的任务数据为例。voidprint_task_data(void*data){intrecv_data*(int*)data;printf(the salary is %d\n,recv_data);}2.4 给任务队列总加入任务加入任务就是将单节点插入到任务队列中。/** * brief 向线程池的任务队列中添加一个任务并唤醒一个工作线程去执行它 * param manager 线程池管理者指针 * param sin_task 待添加的任务节点由调用者分配内存并设置好回调函数和数据 * return 0 成功-1 参数错误-2 线程池正在关闭中无法添加任务 */intthreadpool_addtask(Manager*manager,Task*sin_task){// 1. 参数校验管理者、任务以及任务队列都必须存在if(!manager||!sin_task||!manager-task_queue)return-1;// 2. 加锁后续对任务队列和关闭标志的修改需要互斥保护pthread_mutex_lock(manager-mutex);// 3. 检查关闭标志如果线程池正在关闭不能再添加新任务避免任务泄漏if(manager-shutdown){pthread_mutex_unlock(manager-mutex);return-2;}// 4. 将任务节点插入到任务队列尾部FIFO 入队sin_task-nextNULL;// 新任务成为新的尾部next 置空if(manager-task_queue-tail)// 队列非空{manager-task_queue-tail-nextsin_task;// 当前尾节点指向新任务manager-task_queue-tailsin_task;// 更新尾指针}else// 队列为空新任务同时是头也是尾{manager-task_queue-headsin_task;manager-task_queue-tailsin_task;}// 5. 唤醒一个正在等待任务的工作线程// 注意这里使用 signal 而不是 broadcast因为只增加了一个任务// 唤醒一个空闲线程即可避免无效的“惊群效应”pthread_cond_signal(manager-cond);// 6. 解锁任务入队和唤醒操作已完成pthread_mutex_unlock(manager-mutex);return0;}2.5 线程池的销毁线程池的销毁需要释放之前每个分批工作队列的空间通过pthread_join销毁线程intthreadpool_destroy(Manager*manager){if(!manager)return-1;pthread_mutex_lock(manager-mutex);manager-shutdown1;// Wake up all worker threads so they can drain remaining tasks and exit.pthread_cond_broadcast(manager-cond);pthread_mutex_unlock(manager-mutex);Worker*sin_workermanager-worker_queue-head;while(sin_worker){Worker*worker_nextsin_worker-next;pthread_join(sin_worker-thread_id,NULL);free(sin_worker);sin_workerworker_next;}pthread_mutex_destroy(manager-mutex);pthread_cond_destroy(manager-cond);free(manager-task_queue);free(manager-worker_queue);manager-task_queueNULL;manager-worker_queueNULL;return0;}