线程池的原理与C语言实现 - IT评测·应用市场-qidao123.com

// 任务结点单向链表的节点，类型
struct task
{
void *(*do_task)(void *arg); // 任务函数指针指向线程要执行的任务格式是固定的
void *arg; // 需要传递给任务的参数，如果不需要，则NULL
struct task *next; // 指向下一个任务结点的指针
};

复制代码

// 初始化线程池 pool线程池指针 threads_number 初始化线程的个数
bool init_pool(thread_pool *pool, unsigned int threads_number)
{
// 初始化互斥锁
pthread_mutex_init(&pool->lock, NULL);
// 初始化条件量
pthread_cond_init(&pool->cond, NULL);
// 销毁标志设置线程池为未关闭状态
pool->shutdown = false; // 不销毁
// 给任务链表的节点申请堆内存
pool->task_list = malloc(sizeof(struct task));
// 申请堆内存，用于存储创建出来的线程的ID
pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);
// 错误处理，对malloc进行错误处理
if (pool->task_list == NULL || pool->tids == NULL)
{
perror("分配内存错误");
return false;
}
// 对任务链表中的节点的指针域进行初始化
pool->task_list->next = NULL;
// 设置线程池中处于等待状态的任务数量最大值
pool->max_waiting_tasks = MAX_WAITING_TASKS;
// 设置等待线程处理的任务的数量为0，说明现在没有任务
pool->waiting_tasks = 0;
// 设置线程池中活跃的线程的数量
pool->active_threads = threads_number;
int i;
// 循环创建活跃线程
for (i = 0; i < pool->active_threads; i++)
{
// 创建线程把线程的ID存储在申请的堆内存
if (pthread_create(&((pool->tids)[i]), NULL,
routine, (void *)pool) != 0)
{
perror("创建线程错误");
return false;
}
}
return true;
}

复制代码

// 向线程池的任务链表中添加任务
bool add_task(thread_pool *pool,
void *(*do_task)(void *arg), void *arg)
{
// 给任务链表节点申请内存
struct task *new_task = malloc(sizeof(struct task));
if (new_task == NULL) // 检查内存分配是否成功
{
perror("申请内存错误");
return false;
}
new_task->do_task = do_task; // 设置任务函数指针
new_task->arg = arg; // 设置任务参数
new_task->next = NULL; // 指针域设置为NULL 初始化任务的下一个指针
//============ LOCK =============//
pthread_mutex_lock(&pool->lock); // 加锁，保护共享资源
//===============================//
// 说明要处理的任务的数量大于能处理的任务数量
if (pool->waiting_tasks >= MAX_WAITING_TASKS) // 检查等待任务是否超过最大值
{
pthread_mutex_unlock(&pool->lock); // 解锁
fprintf(stderr, "任务太多.\n"); // 打印错误信息
free(new_task); // 释放新任务内存
return false;
}
struct task *tmp = pool->task_list; // 获取任务链表头
// 遍历链表，找到单向链表的尾节点
while (tmp->next != NULL)
tmp = tmp->next;
// 把新的要处理的任务插入到链表的尾部尾插
tmp->next = new_task;
// 要处理的任务的数量+1 (等待任务数量+1)
pool->waiting_tasks++;
//=========== UNLOCK ============//
pthread_mutex_unlock(&pool->lock); // 解锁
//===============================//
// 唤醒第一个处于阻塞队列中的线程
pthread_cond_signal(&pool->cond);
return true;
}

复制代码

// 向线程池加入新线程
int add_thread(thread_pool *pool, unsigned additional_threads)
{
// 判断需要添加的新线程的数量是否为0 如果没有要添加的线程，直接返回
if (additional_threads == 0)
return 0;
// 计算线程池中总线程的数量
unsigned total_threads =
pool->active_threads + additional_threads;
int i, actual_increment = 0; // 初始化计数器
// 循环创建新线程
for (i = pool->active_threads; i < total_threads && i < MAX_ACTIVE_THREADS; i++)
{
// 创建新线程
if (pthread_create(&((pool->tids)[i]),
NULL, routine, (void *)pool) != 0)
{
perror("增加活跃线程错误"); // 打印错误信息
// 如果没有成功创建任何线程，返回错误
if (actual_increment == 0)
return -1;
break; // 退出循环
}
actual_increment++; // 增加计数器
}
// 记录此时线程池中活跃线程的总数
pool->active_threads += actual_increment; // 更新活跃线程数
return actual_increment; // 返回实际增加的线程数
}

复制代码

// 从线程池中删除线程
int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
if (removing_threads == 0)
return pool->active_threads; // 如果没有要删除的线程，直接返回
int remaining_threads = pool->active_threads - removing_threads; // 计算剩余线程数
remaining_threads = remaining_threads > 0 ? remaining_threads : 1; // 确保至少有一个线程
int i;
for (i = pool->active_threads - 1; i > remaining_threads - 1; i--) // 循环取消线程
{
errno = pthread_cancel(pool->tids[i]); // 取消线程
if (errno != 0) // 检查取消是否成功
break;
}
if (i == pool->active_threads - 1) // 如果没有成功取消任何线程，返回错误
return -1;
else
{
pool->active_threads = i + 1; // 更新活跃线程数
return i + 1; // 返回剩余线程数
}
}

复制代码

// 销毁线程池
bool destroy_pool(thread_pool *pool)
{
// 1,激活所有线程设置关闭标志
pool->shutdown = true;
pthread_cond_broadcast(&pool->cond); // 唤醒所有等待中的线程
// 2, 等待线程们执行完毕
int i;
for (i = 0; i < pool->active_threads; i++) // 循环等待所有线程退出
{
/**
* pthread_join(pool->tids[i], NULL) 的作用是等待线程池中第 i 个线程终止，并清理其相关资源。通过这种方式，可以确保在销毁线程池时，所有线程都已经安全地终止。
* pthread_join 是 POSIX 线程库中的一个函数，用于等待一个线程的终止。它的功能类似于进程中的 wait 系统调用。
*/
errno = pthread_join(pool->tids[i], NULL); // 等待线程退出
if (errno != 0) // 检查等待是否成功
{
printf("join tids[%d] error: %s\n",
i, strerror(errno)); // 打印错误信息
}
else
printf("[%u] is joined\n", (unsigned)pool->tids[i]); // 打印线程退出信息
}
// 3, 销毁线程池
free(pool->task_list); // 释放任务链表内存
free(pool->tids); // 释放线程ID数组内存
free(pool); // 释放线程池结构体内存
return true;
}

复制代码

#include "thread_pool.h" // 包含线程池头文件
// 任务函数, 打印一次线程任务信息,并等待n秒,模拟真正的线程任务
void *mytask(void *arg)
{
int n = (int)arg; // 要执行的秒数将参数转换为整数, 强制转换才能使用
/**
* %u：无符号整数（unsigned int）
* pthread_self()：这是一个 POSIX 线程库函数，返回调用它的线程的线程 ID。
* __FUNCTION__：这是一个预定义的宏，扩展为当前函数的名称。它在调试和日志记录时非常有用，可以显示当前正在执行的函数名。
*/
printf("[%u][%s] ==>工作将会在这里被执行 %d 秒...\n",
(unsigned)pthread_self(), __FUNCTION__, n); // 打印任务开始信息
sleep(n);
printf("[%u][%s] ==> 工作完毕!\n",
(unsigned)pthread_self(), __FUNCTION__); // 打印任务完成信息
return NULL;
}
// 计时函数
void *count_time(void *arg)
{
int i = 0; // 初始化计数器
while (1)
{
sleep(1);
printf("sec: %d\n", ++i); // 打印经过的秒数
}
}
int main(void)
{
pthread_t a; // 定义一个线程ID
pthread_create(&a, NULL, count_time, NULL); // 创建计时线程
// 1, 初始化线程池
thread_pool *pool = malloc(sizeof(thread_pool)); // 分配内存给线程池管理结构体
init_pool(pool, 2); // 初始化线程池，创建2个线程
// 2, 添加任务
printf("向线程池中投送3个任务...\n");
/**
* rand() 是 C 标准库函数，定义在 <stdlib.h> 头文件中。它返回一个伪随机数，
* 范围在 0 到 RAND_MAX 之间，RAND_MAX 是一个宏，通常定义为 32767。
*
* rand() % 10 的结果是 rand() 产生的随机数对 10 取模的结果，也就是说，它会返回一个 0 到 9 之间的整数（包括 0 和 9）
*
* 线程函数和任务函数通常需要一个 void * 类型的参数，以便能够传递任意类型的数据。在这种情况下，任务函数 mytask 需要一个 void * 类型的参数。
*/
add_task(pool, mytask, (void *)(rand() % 10));
add_task(pool, mytask, (void *)(rand() % 10));
add_task(pool, mytask, (void *)(rand() % 10));
// 3, 检查活跃线程数量
printf("当前活跃的线程数量: %d\n",
remove_thread(pool, 0)); // 打印当前活跃线程数
sleep(9); // 等待9秒
// 4, 添加更多任务
printf("向线程池中投送2个任务...\n"); // 打印信息
add_task(pool, mytask, (void *)(rand() % 10));
add_task(pool, mytask, (void *)(rand() % 10));
// 5, 添加线程
add_thread(pool, 2); // 添加2个线程
sleep(5); // 等待5秒
// 6, 删除线程
printf("从线程池中删除3个活跃线程, "
"当前线程数量: %d\n",
remove_thread(pool, 3));
// 7, 销毁线程池
destroy_pool(pool); // 销毁线程池
return 0; // 程序正常结束
}

复制代码

#ifndef _THREAD_POOL_H_#define _THREAD_POOL_H_#include // 尺度输入输出库#include // 布尔类型库#include // UNIX 尺度库，包含 sleep 函数#include // 尺度库，包含 malloc 和 free 函数#include // 字符串处理惩罚库#include // 字符串处理惩罚库#include // 错误号库#include // POSIX 线程库#define MAX_WAITING_TASKS 1000 // 处于等待状态的任务数量最大为1000#define MAX_ACTIVE_THREADS 20 // 活跃线程的最大数量, 但该数量最佳应该==CPU一次性可执行的线程数量, 例如6核12线程, 则为12/*************第一步: 构建任务布局体******************/// 任务结点单向链表的节点，类型
struct task
{
void *(*do_task)(void *arg); // 任务函数指针指向线程要执行的任务格式是固定的
void *arg; // 需要传递给任务的参数，如果不需要，则NULL
struct task *next; // 指向下一个任务结点的指针
};// 线程池的管理布局体typedef struct thread_pool{ pthread_mutex_t lock; // 互斥锁, 用于掩护任务队列 pthread_cond_t cond; // 条件量, 代表任务队列中任务个数的变革---如果主线程向队列投放任务, 则可以通过条件变量来唤醒哪些睡着了的线程 bool shutdown; // 是否必要销毁线程池, 控制线程退出, 进而销毁整个线程池 struct task *task_list; // 用于存储任务的链表, 任务队列刚开始没有任何任务, 是一个具有头节点的空链队列 pthread_t *tids; // 用于记录线程池中线程的ID unsigned max_waiting_tasks; // 线程池中处于等待状态的任务数量最大值 unsigned waiting_tasks; // 处于等待状态的线程数量 unsigned active_threads; // 正在活跃的线程数量} thread_pool;// 初始化线程池bool init_pool(thread_pool *pool, unsigned int threads_number);// 向线程池中添加任务bool add_task(thread_pool *pool, void *(*do_task)(void *arg), void *task);// 先线程池中添加线程int add_thread(thread_pool *pool, unsigned int additional_threads_number);// 从线程池中删除线程int remove_thread(thread_pool *pool, unsigned int removing_threads_number);// 销毁线程池bool destroy_pool(thread_pool *pool);// 任务函数线程例程void *routine(void *arg);#endif

复制代码

#include "thread_pool.h" // 包含线程池头文件// 线程取消处理惩罚函数，确保线程取消时解锁互斥锁void handler(void *arg){ printf("[%u] 竣事了.\n", (unsigned)pthread_self()); // 打印线程竣事信息 pthread_mutex_unlock((pthread_mutex_t *)arg); // 解锁互斥锁}// 线程执行的任务函数void *routine(void *arg){ // 调试 #ifdef DEBUG printf("[%u] is started.\n", (unsigned)pthread_self()); // 打印线程开始信息 #endif // 把必要传递给线程任务的参数进行备份 thread_pool *pool = (thread_pool *)arg; // 将传入的参数转换为线程池指针 struct task *p; // 定义一个任务指针 while (1) // 无限循环，持续处理惩罚任务 { /* ** push a cleanup functon handler(), make sure that ** the calling thread will release the mutex properly ** even if it is cancelled during holding the mutex. ** ** NOTE: ** pthread_cleanup_push() is a macro which includes a ** loop in it, so if the specified field of codes that ** paired within pthread_cleanup_push() and pthread_ ** cleanup_pop() use 'break' may NOT break out of the ** truely loop but break out of these two macros. ** see line 61 below. */ /* * 注意: * 推送一个整理函数handler()，确保调用线程将精确释放互斥量，即使它在持有互斥量期间被取消。 * * pthread_cleanup_push()是一个宏，其中包含一个循环， * 所以如果在pthread_cleanup_push()和pthread_ cleanup_pop()中配对的代码的指定字段使用` break `可能不会跳出真正的循环， * 而是跳出这两个宏。参见下面的第61行。 */ //================================================// pthread_cleanup_push(handler, (void *)&pool->lock); // 注册取消处理惩罚函数 pthread_mutex_lock(&pool->lock); // 加锁，掩护共享资源 //================================================// // 1,如果没有任务且线程池未关闭，则等待 while (pool->waiting_tasks == 0 && !pool->shutdown) { pthread_cond_wait(&pool->cond, &pool->lock); // 等待条件变量 } // 2, 如果没有任务且线程池已关闭，则退出 if (pool->waiting_tasks == 0 && pool->shutdown == true) { pthread_mutex_unlock(&pool->lock); // 解锁 pthread_exit(NULL); // CANNOT use 'break'; 退出线程 } // 3, 有任务则取出任务 p = pool->task_list->next; // 获取第一个任务 pool->task_list->next = p->next; // 将任务从链表中移除 pool->waiting_tasks--; // 减少等待任务计数 //================================================// pthread_mutex_unlock(&pool->lock); // 解锁 pthread_cleanup_pop(0); // 取消注册的取消处理惩罚函数 //================================================// pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); // 禁止线程取消 (p->do_task)(p->arg); // 执行任务 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); // 允许线程取消 free(p); // 释放任务内存 } pthread_exit(NULL); // 退出线程}// 初始化线程池 pool线程池指针 threads_number 初始化线程的个数bool init_pool(thread_pool *pool, unsigned int threads_number){ // 初始化互斥锁 pthread_mutex_init(&pool->lock, NULL); // 初始化条件量 pthread_cond_init(&pool->cond, NULL); // 销毁标记设置线程池为未关闭状态 pool->shutdown = false; // 不销毁 // 给任务链表的节点申请堆内存 pool->task_list = malloc(sizeof(struct task)); // 申请堆内存，用于存储创建出来的线程的ID pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS); // 错误处理惩罚，对malloc进行错误处理惩罚 if (pool->task_list == NULL || pool->tids == NULL) { perror("分配内存错误"); return false; } // 对任务链表中的节点的指针域进行初始化 pool->task_list->next = NULL; // 设置线程池中处于等待状态的任务数量最大值 pool->max_waiting_tasks = MAX_WAITING_TASKS; // 设置等待线程处理惩罚的任务的数量为0，阐明现在没有任务 pool->waiting_tasks = 0; // 设置线程池中活跃的线程的数量 pool->active_threads = threads_number; int i; // 循环创建活跃线程 for (i = 0; i < pool->active_threads; i++) { // 创建线程把线程的ID存储在申请的堆内存 if (pthread_create(&((pool->tids)[i]), NULL, routine, (void *)pool) != 0) { perror("创建线程错误"); return false; } // 用于调试 #ifdef DEBUG printf("[%u]:[%s] ==> tids[%d]: [%u] is created.\n", (unsigned)pthread_self(), __FUNCTION__, i, (unsigned)pool->tids[i]); // 打印线程创建信息 #endif } return true;}// 向线程池的任务链表中添加任务bool add_task(thread_pool *pool, void *(*do_task)(void *arg), void *arg){ // 给任务链表节点申请内存 struct task *new_task = malloc(sizeof(struct task)); if (new_task == NULL) // 检查内存分配是否成功 { perror("申请内存错误"); return false; } new_task->do_task = do_task; // 设置任务函数指针 new_task->arg = arg; // 设置任务参数 new_task->next = NULL; // 指针域设置为NULL 初始化任务的下一个指针 //============ LOCK =============// pthread_mutex_lock(&pool->lock); // 加锁，掩护共享资源 //===============================// // 阐明要处理惩罚的任务的数量大于能处理惩罚的任务数量 if (pool->waiting_tasks >= MAX_WAITING_TASKS) // 检查等待任务是否超过最大值 { pthread_mutex_unlock(&pool->lock); // 解锁 fprintf(stderr, "任务太多.\n"); // 打印错误信息 free(new_task); // 释放新任务内存 return false; } struct task *tmp = pool->task_list; // 获取任务链表头 // 遍历链表，找到单向链表的尾节点 while (tmp->next != NULL) tmp = tmp->next; // 把新的要处理惩罚的任务插入到链表的尾部尾插 tmp->next = new_task; // 要处理惩罚的任务的数量+1 (等待任务数量+1) pool->waiting_tasks++; //=========== UNLOCK ============// pthread_mutex_unlock(&pool->lock); // 解锁 //===============================// // 调试 #ifdef DEBUG printf("[%u][%s] ==> a new task has been added.\n", (unsigned)pthread_self(), __FUNCTION__); // 打印任务添加信息 #endif // 唤醒第一个处于壅闭队列中的线程 pthread_cond_signal(&pool->cond); return true;}// 向线程池参加新线程int add_thread(thread_pool *pool, unsigned additional_threads){ // 判断必要添加的新线程的数量是否为0 如果没有要添加的线程，直接返回 if (additional_threads == 0) return 0; // 计算线程池中总线程的数量 unsigned total_threads = pool->active_threads + additional_threads; int i, actual_increment = 0; // 初始化计数器 // 循环创建新线程 for (i = pool->active_threads; i < total_threads && i < MAX_ACTIVE_THREADS; i++) { // 创建新线程 if (pthread_create(&((pool->tids)[i]), NULL, routine, (void *)pool) != 0) { perror("增加活跃线程错误"); // 打印错误信息 // 如果没有成功创建任何线程，返回错误 if (actual_increment == 0) return -1; break; // 退出循环 } actual_increment++; // 增加计数器 #ifdef DEBUG printf("[%u]:[%s] ==> tids[%d]: [%u] is created.\n", (unsigned)pthread_self(), __FUNCTION__, i, (unsigned)pool->tids[i]); // 打印线程创建信息 #endif } // 记录此时线程池中活跃线程的总数 pool->active_threads += actual_increment; // 更新活跃线程数 return actual_increment; // 返回实际增加的线程数}// 从线程池中删除线程int remove_thread(thread_pool *pool, unsigned int removing_threads){ if (removing_threads == 0) return pool->active_threads; // 如果没有要删除的线程，直接返回 int remaining_threads = pool->active_threads - removing_threads; // 计算剩余线程数 remaining_threads = remaining_threads > 0 ? remaining_threads : 1; // 确保至少有一个线程 int i; for (i = pool->active_threads - 1; i > remaining_threads - 1; i--) // 循环取消线程 { errno = pthread_cancel(pool->tids[i]); // 取消线程 if (errno != 0) // 检查取消是否成功 break; #ifdef DEBUG printf("[%u]:[%s] ==> cancelling tids[%d]: [%u]...\n", (unsigned)pthread_self(), __FUNCTION__, i, (unsigned)pool->tids[i]); // 打印线程取消信息 #endif } if (i == pool->active_threads - 1) // 如果没有成功取消任何线程，返回错误 return -1; else { pool->active_threads = i + 1; // 更新活跃线程数 return i + 1; // 返回剩余线程数 }}// 销毁线程池
bool destroy_pool(thread_pool *pool)
{
// 1,激活所有线程设置关闭标志
pool->shutdown = true;
pthread_cond_broadcast(&pool->cond); // 唤醒所有等待中的线程
// 2, 等待线程们执行完毕
int i;
for (i = 0; i < pool->active_threads; i++) // 循环等待所有线程退出
{
/**
* pthread_join(pool->tids[i], NULL) 的作用是等待线程池中第 i 个线程终止，并清理其相关资源。通过这种方式，可以确保在销毁线程池时，所有线程都已经安全地终止。
* pthread_join 是 POSIX 线程库中的一个函数，用于等待一个线程的终止。它的功能类似于进程中的 wait 系统调用。
*/
errno = pthread_join(pool->tids[i], NULL); // 等待线程退出
if (errno != 0) // 检查等待是否成功
{
printf("join tids[%d] error: %s\n",
i, strerror(errno)); // 打印错误信息
}
else
printf("[%u] is joined\n", (unsigned)pool->tids[i]); // 打印线程退出信息
}
// 3, 销毁线程池
free(pool->task_list); // 释放任务链表内存
free(pool->tids); // 释放线程ID数组内存
free(pool); // 释放线程池结构体内存
return true;
}

复制代码