11.5 Linux_线程_线程池

🕗 发布于 2024-10-07 21:41 linux

概述

什么是线程池：

线程池就是线程的集合，里面存放了一系列的线程。

线程池使用于需要大量创建和销毁线程的情况，使用线程池比单个线程的创建和销毁速度要快。

线程池的结构：

任务队列：任务队列是任务的生产者，存储需要处理的任务，工作线程会处理这些任务。

线程池：线程池中存放着工作线程，它是任务队列任务的消费者，等待新任务的信号。

固定大小线程池的实现：

1、创建线程池的基本结构：任务队列结构体、线程池结构体

2、线程池初始化：创建线程池结构、互斥锁和条件变量初始化、创建n个工作线程

3、线程池添加任务：判断是否有空闲线程、给任务队列添加节点、给工作线程发送信号

4、实现工作线程：等待任务信号、从任务队列中取出任务、执行任务

5、线程池销毁：删除任务队列、互斥锁、条件变量、线程池

线程池的实现

1、结构体实现

1.1 任务队列结构体

任务被抽象为一个函数，队列需要一个指针指向下一个节点。

具体结构体声明如下：

//任务队列
typedef struct Task{
void* (*func)(void* arg);//任务处理函数
void* arg;               //任务处理函数的参数
struct Task* pNext;      //链表指针
}Task;

1.2 线程池结构体

任务是一个临界资源，同一时刻只允许有一个线程处理该任务，所以需要一个互斥量。

任务和工作线程为生产者和消费者的关系，所以需要一个条件变量。

线程池中有多个工作线程，所以需要一个线程TID数组

线程与任务之间需要交互，所以需要当前正在工作线程的个数、当前需要处理的任务

具体结构体声明如下：

//线程池
#define POOL_NUM 5 
typedef struct ThreadPool{
pthread_mutex_t taskLock;    //互斥锁
pthread_cond_t newTask;      //条件变量

pthread_t tid[POOL_NUM];     //线程TID
Task* queue_head;            //任务队列头部
Task* queue_tail;     //任务队列尾部
int busyWork;                //当前正在工作的线程数量
}ThreadPool;
ThreadPool* pPool;

2、创建和销毁线程池

2.1 初始化线程池

线程池初始化就是申请线程池空间，并初始化互斥量、条件变量、任务队列、工作线程数量、创建工作线程。

具体代码实现如下：

/*
 * pool_init:线程池初始化
 * @ret -1--err 0--success
 */
int pool_init(void){

//1.申请空间
if((pPool = malloc(sizeof(ThreadPool))) == NULL){
printf("malloc err\n");
return -1;
}
//2.初始化
pthread_mutex_init(&(pPool->taskLock),NULL); //互斥量初始化
pthread_cond_init(&(pPool->newTask),NULL); //条件变量初始化
pPool->busyWork = 0;//当前工作的线程=0
if((pPool->queue_head = malloc(sizeof(Task))) == NULL){ //任务队列初始化
printf("malloc err\n");
return -1;
}
pPool->queue_tail = pPool->queue_head;
pPool->queue_head->pNext = NULL;
for(int i=0;i<POOL_NUM;i++){ //创建工作线程
if(pthread_create(&(pPool->tid[i]),NULL,workThread,NULL) != 0){
perror("pthread_create");
return -1;//这里没有做清理工作
}
}
return 0;
}

2.2 销毁线程池

销毁线程池就是释放创建时申请的空间。销毁顺序为：任务队列、互斥量和条件变量、线程池

具体代码实现如下：

/*
 * pool_destroy:销毁线程池
 * */
void pool_destroy(void){
Task* pTask = NULL;
//1.销毁任务队列
while(pPool->queue_head!=NULL){
pTask = pPool->queue_head->pNext;
free(pPool->queue_head);
pPool->queue_head = pTask;
}
//2.销毁互斥锁、条件变量
pthread_mutex_destroy(&(pPool->taskLock));
pthread_cond_destroy(&(pPool->newTask));
//3.销毁线程池
free(pPool);
}

3、向线程池中添加任务(生产者)

该函数首先是生产者模型：加锁互斥量->生成资源->发送信号->解锁互斥量

除此之外，还需在生产资源之前判断是否存在空闲的线程，如果不存在则需等待一段时间再生产资源，这一步的目的是为了防止生产资源后发送的信号没有被工作线程接收，导致资源的丢失。

具体代码实现如下：

/*
 * pool_add_task:向线程池中添加任务
 * param pfun:任务处理函数
 * param arg:任务处理函数的参数
 * */
void pool_add_task(void*(pfun)(void*),void* arg){
Task* pTask = NULL;
//生产者代码框架
pthread_mutex_lock(&(pPool->taskLock)); //上锁互斥量
//判断当前是否有空闲的工作线程,如果不判断会导致条件变量信号丢失问题
while(pPool->busyWork > POOL_NUM){
pthread_mutex_unlock(&(pPool->taskLock));
usleep(1000);
pthread_mutex_lock(&(pPool->taskLock));
}
if((pTask = malloc(sizeof(Task))) == NULL){ //生产资源
printf("malloc err\n");
return;
}
pTask->func = pfun;
pTask->arg = arg;
pTask->pNext = NULL;
pPool->queue_tail->pNext = pTask;
pPool->queue_tail = pTask;
pPool->busyWork++; //资源生产结束,代表将有一个线程会工作
pthread_cond_signal(&(pPool->newTask)); //发送信号
pthread_mutex_unlock(&(pPool->taskLock)); //解锁互斥量
}

4、从线程池中处理任务(消费者)

该函数首先是消费者模型：上锁互斥量->等待信号->获取资源->解锁互斥量

除此之外，这里的获取资源指的是将任务结构体获取到，而不是直接处理完成任务。任务的处理与生产者与消费者模型无关，所以可以放在解锁互斥量后面，这一步的目的是为了尽快解锁互斥量，让其他线程可以尽快处理新来的任务。

具体代码实现如下：

/*
 * workThread:工作线程
 * */
void* workThread(void* arg){
Task* pTask = NULL;
while(1){
//消费者代码框架
pthread_mutex_lock(&(pPool->taskLock)); //上锁互斥量
while(pPool->queue_head->pNext == NULL){    //等待资源
pthread_cond_wait(&(pPool->newTask),&(pPool->taskLock));
}
pTask = pPool->queue_head->pNext; //使用资源
pPool->queue_head->pNext = pTask->pNext;
if(pPool->queue_head->pNext == NULL){
pPool->queue_tail = pPool->queue_head;
}
pthread_mutex_unlock(&(pPool->taskLock));   //解锁互斥量
//资源中的任务与消费者无关，因此可以放在互斥量外
pTask->func(pTask->arg); //执行相应任务函数
free(pTask);
pPool->busyWork--; //当前线程工作完成
}
}

线程池完整代码

代码功能：

在main函数中不断的生产资源，生产资源后，工作线程会自动的处理资源。

具体代码实现如下：

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>

//任务队列
typedef struct Task{
void* (*func)(void* arg);//任务处理函数
void* arg;               //任务处理函数的参数
struct Task* pNext;      //链表指针
}Task;
//线程池
#define POOL_NUM 5 
typedef struct ThreadPool{
pthread_mutex_t taskLock;    //互斥锁
pthread_cond_t newTask;      //条件变量

pthread_t tid[POOL_NUM];     //线程TID
Task* queue_head;            //任务队列头部
Task* queue_tail;     //任务队列尾部
int busyWork;                //当前正在工作的线程数量
}ThreadPool;
ThreadPool* pPool;

int pool_init(void);
void* workThread(void* arg);
void pool_add_task(void* (pfun)(void*),void* arg);
void* task1(void* arg);
void pool_destroy(void);

int main(){

int tmp = 1;
pool_init();
while(1){
pool_add_task(task1,(void*)tmp);//注意:这里tmp是常变化的值，不能传入地址&tmp
tmp++;
}

return 0;
}
/*
 * pool_destroy:销毁线程池
 * */
void pool_destroy(void){
Task* pTask = NULL;
//1.销毁任务队列
while(pPool->queue_head!=NULL){
pTask = pPool->queue_head->pNext;
free(pPool->queue_head);
pPool->queue_head = pTask;
}
//2.销毁互斥锁、条件变量
pthread_mutex_destroy(&(pPool->taskLock));
pthread_cond_destroy(&(pPool->newTask));
//3.销毁线程池
free(pPool);
}

/*
 * task1:模拟任务
 * */
void* task1(void* arg){
printf("%ld\n",((long int)arg));
sleep(1);
return NULL;
}

/*
 * pool_add_task:向线程池中添加任务
 * param pfun:任务处理函数
 * param arg:任务处理函数的参数
 * */
void pool_add_task(void*(pfun)(void*),void* arg){
Task* pTask = NULL;
//生产者代码框架
pthread_mutex_lock(&(pPool->taskLock)); //上锁互斥量
//判断当前是否有空闲的工作线程,如果不判断会导致条件变量信号丢失问题
while(pPool->busyWork > POOL_NUM){
pthread_mutex_unlock(&(pPool->taskLock));
usleep(1000);
pthread_mutex_lock(&(pPool->taskLock));
}
if((pTask = malloc(sizeof(Task))) == NULL){ //生产资源
printf("malloc err\n");
return;
}
pTask->func = pfun;
pTask->arg = arg;
pTask->pNext = NULL;
pPool->queue_tail->pNext = pTask;
pPool->queue_tail = pTask;
pPool->busyWork++; //资源生产结束,代表将有一个线程会工作
pthread_cond_signal(&(pPool->newTask)); //发送信号
pthread_mutex_unlock(&(pPool->taskLock)); //解锁互斥量
}

/*
 * workThread:工作线程
 * */
void* workThread(void* arg){
Task* pTask = NULL;
while(1){
//消费者代码框架
pthread_mutex_lock(&(pPool->taskLock)); //上锁互斥量
while(pPool->queue_head->pNext == NULL){    //等待资源
pthread_cond_wait(&(pPool->newTask),&(pPool->taskLock));
}
pTask = pPool->queue_head->pNext; //使用资源
pPool->queue_head->pNext = pTask->pNext;
if(pPool->queue_head->pNext == NULL){
pPool->queue_tail = pPool->queue_head;
}
pthread_mutex_unlock(&(pPool->taskLock));   //解锁互斥量
//资源中的任务与消费者无关，因此可以放在互斥量外
pTask->func(pTask->arg); //执行相应任务函数
free(pTask);
pPool->busyWork--; //当前线程工作完成
}
}

/*
 * pool_init:线程池初始化
 * @ret -1--err 0--success
 */
int pool_init(void){

//1.申请空间
if((pPool = malloc(sizeof(ThreadPool))) == NULL){
printf("malloc err\n");
return -1;
}
//2.初始化
pthread_mutex_init(&(pPool->taskLock),NULL); //互斥量初始化
pthread_cond_init(&(pPool->newTask),NULL); //条件变量初始化
pPool->busyWork = 0;//当前工作的线程=0
if((pPool->queue_head = malloc(sizeof(Task))) == NULL){ //任务队列初始化
printf("malloc err\n");
return -1;
}
pPool->queue_tail = pPool->queue_head;
pPool->queue_head->pNext = NULL;
for(int i=0;i<POOL_NUM;i++){ //创建工作线程
if(pthread_create(&(pPool->tid[i]),NULL,workThread,NULL) != 0){
perror("pthread_create");
return -1;//这里没有做清理工作
}
}
return 0;
}

原文地址：https://blog.csdn.net/Fresh_man111/article/details/142695691

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：【笔记】选择题笔记+数据结构笔记
下一篇：如何方便地打出「」和『』

无人机集群路径规划：四种优化算法（BKA、CO、PSO、PIO）求解无人机集群路径规划，提供MATLAB代码
无人机三维路径规划是指在三维空间中为无人机规划一条合理的飞行路径，使其能够安全、高效地完成任务。路径规划是无人机自主飞行的关键技术之一，它可以通过算法和模型来确定无人机的航迹，以避开障碍物、优化飞行时
阅读更多2024-10-08
什么是变阻器？
通过合理地设置变阻器的电阻值，可以将电源电压分成所需的多个电压值，从而满足电路的需求。
阅读更多2024-10-08
malloc分配内存的两种方式——brk、mmap
从操作系统角度来看，进程分配内存有两种方式，分别由两个完成。
阅读更多2024-10-08
自动生成实体类，mapper类，mapper.xml文件
自动生成实体类，mapper类，mapper.xml文件
阅读更多2024-10-08
tcpdump-arm平台移植
tcpdump-arm平台移植
阅读更多2024-10-08
JavaEE: 深入解析HTTP协议的奥秘(3)
解决方案: 同时开发出多个不同版本的页面(有的只有文字图片,有的包含复杂多媒体),判断 UA 字段,通过 UA 获取到用户的浏览器信息和操作系统信息,这样就可以判定当前用户的浏览器版本都支持哪些特性,
阅读更多2024-10-08
C++多态常见问题
重写指的是在继承关系中子类重新定义继承来的虚函数重写的函数名\参数列表\返回值类型必须和父类一致被继承的函数必须是虚函数可以指定父类显示调用基类的被重写函数属于动态多态通过重写子类中的虚函数
阅读更多2024-10-08
数据库增，删，改常用语句
select语句中只允许被分组的列和为每个分组返回的一个值的表达试，例如用一个列名作为参数的聚合函数。说明：在strdents表和score表中查询满足on条件的行，条件为score表的strdent
阅读更多2024-10-08
深搜剪枝优化
！！！！！
阅读更多2024-10-08
ssm题库管理系统的设计与实现
需要定制毕设请私聊，海量题目可选，你选题目我出功能
阅读更多2024-10-08

11.5 Linux_线程_线程池

概述

线程池的实现

1、结构体实现

1.1 任务队列结构体

1.2 线程池结构体

2、创建和销毁线程池

2.1 初始化线程池

2.2 销毁线程池

3、向线程池中添加任务(生产者)

4、从线程池中处理任务(消费者)

线程池完整代码

相关文章