跳到主要内容

单向链表创建线程池

· 阅读需 9 分钟
📺尚宇

使用单向链表实现线程池的接口设计

1. 基础知识​

关于链表和线程的使用方法可以查看我的技术笔记：数据结构 和 系统编程 ，在这两个页面都可以找到链表和线程的一些基础知识和使用示例。

在 Linux 系统中，线程池是一种用于管理和调度线程的机制。它由一组可以复用的线程组成，这些线程被保存在线程池中，并等待分配任务。

当一个新的任务到达时，线程池会从其中选择一个线程来执行该任务。如果所有线程都正在执行任务，则新任务将被放在队列中等待，直到有线程可用。

使用线程池的优点包括：

  • 可以更有效地利用 CPU 资源，因为线程可以在任务之间复用。
  • 可以更容易地管理线程，因为所有线程都在线程池中。
  • 可以更容易地维护线程的数量，因为可以限制线程池的大小。

线程池的实现原理可以参考这篇笔记：线程池。

下面我们来看看怎么通过单向链表跟线程的组合实现线程池。

2. 线程池接口设计​

2.1 任务结构体​

  • 接口文档
原型struct task
功能描述任务节点，包含需要执行的函数及其参数，通过链表连成一个任务队列
成员列表void *(*task)(void *arg);
void *arg;
struct task *next;
备注任务队列最终是形成一条单向链表
  • 代码实现
struct task {	// 任务节点
	void *(*task)(void *arg);
	void *arg;
	struct task *next;
};

2.2 线程池结构体​

  • 接口文档
原型thread_pool
功能描述包含线程池的所有信息
成员列表pthread_mutex_t lock; // 互斥锁，保护任务队列
pthread_cond_t cond; // 条件变量，同步所有线程
struct task *task_list; // 任务链队列指针
pthread_t *tids; // 线程ID存放位置
unsigned waiting_tasks; // 任务链队列中等待的任务个数
unsigned active_threads;// 当前活跃线程个数
bool shutdown; // 线程池销毁标记
备注活跃线程个数由用户自行定义，但至少包含一条活跃线程
  • 代码实现
typedef struct thread_pool {// 线程池
	pthread_mutex_t lock;	// 互斥锁，保护任务队列
	pthread_cond_t  cond;	// 条件变量，同步所有线程
	struct task *task_list;	// 任务链队列指针
	pthread_t *tids;		// 线程ID存放位置
	unsigned waiting_tasks;	// 任务链队列中等待的任务个数
	unsigned active_threads;// 当前活跃线程个数
	bool shutdown;			// 线程池销毁标记
}thread_pool;

2.3 线程池初始化​

  • 接口文档
原型bool init_pool(thread_pool* pool, unsigned int threads_number);
功能描述创建一个新的线程池，包含threads_number个活跃线程
参数pool: 线程池指针
threads_number: 初始活跃线程个数(大于等于1)
返回值成功返回 true ，失败返回 false
所在头文件thread_pool.h
备注线程池最少线程个数为1
  • 代码实现
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));	// 初始化任务队列
	pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);	// 初始化存放线程ID的数组

	if(pool->task_list == NULL || pool->tids == NULL)
	{
		perror("allocate memory error");
		return false;
	}

	pool->task_list->next = NULL;	// 链表的下一个节点为空

	pool->waiting_tasks = 0;	// 初始化任务队列中等待的任务个数
	pool->active_threads = threads_number;	//初始化活跃线程个数

	int i;
	// 创建指定数目线程
	for(i=0; i<pool->active_threads; i++) {
		if(pthread_create(&((pool->tids)[i]), NULL, routine, (void *)pool) != 0) {
			perror("create threads error");
			return false;
		}
	}

	return true;
}

2.4 投送任务​

  • 接口文档
原型bool add_task(thread_pool *pool, void *(*do_task)(void *arg), void *arg);
功能描述往线程池投送任务
参数pool: 线程池指针
do_task: 投送至线程池的执行例程
arg: 执行例程do_task的参数，若该执行例程不需要参数可设置为NULL
返回值成功返回true，失败返回false
所在头文件thread_pool.h
备注任务队列中最大任务个数为 MAX_WAITING_TASKS
  • 代码实现
bool add_task(thread_pool *pool, void *(*task)(void *arg), void *arg)
{
	struct task *new_task = malloc(sizeof(struct task));	// 新任务结点

	if(new_task == NULL) {
		perror("allocate memory error");
		return false;
	}

	// 初始化任务节点
	new_task->task = task;
	new_task->arg = arg;
	new_task->next = NULL;

	// 加锁、超过最大任务
	pthread_mutex_lock(&pool->lock);
	if(pool->waiting_tasks >= MAX_WAITING_TASKS) {
		pthread_mutex_unlock(&pool->lock);

		fprintf(stderr, "too many tasks.\n");
		free(new_task);

		return false;
	}
	
	struct task *tmp = pool->task_list;
	while(tmp->next != NULL)
		tmp = tmp->next;

	tmp->next = new_task;	// 添加新的任务结点
	pool->waiting_tasks++;

	// 释放互斥锁，并唤醒其中一个阻塞在条件变量上的线程
	pthread_mutex_unlock(&pool->lock);
	pthread_cond_signal(&pool->cond);

	return true;
}

2.5 增加活跃线程​

  • 接口文档
原型int add_thread(thread_pool*pool, unsigned int additional_threads);
功能描述增加线程池中活跃线程的个数
参数pool: 需要增加线程的线程池指针
additional_threads: 新增线程个数
返回值>0: 实际新增线程个数
-1: 失败
所在头文件thread_pool.h
备注增加的活跃线程的数量和之前初始化的线程数量不能大于 MAX_ACTIVE_THREADS
  • 代码实现
int add_thread(thread_pool *pool, unsigned int additional_threads)
{
	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("add threads error");

			if(actual_increment == 0)
				return -1;

			break;
		}
		actual_increment++; 
	}

	// 更新线程池中的活跃线程个数
	pool->active_threads += actual_increment;

	return actual_increment;
}

2.6 删除线程​

  • 接口文档
原型int remove_thread(thread_pool*pool, unsigned int removing_threads);
功能描述删除线程池中活跃线程的个数
参数pool: 需要删除线程的线程池指针
removing_threads: 要删除的线程个数，该参数设置为0时直接返回当前线程池线程总数，对线程池不造成任何其它影响
返回值>0: 当前线程池剩余线程个数
-1:失败
所在头文件thread_pool.h
备注1，线程池至少会存在1条活跃线程
2，如果被删除的线程正在执行任务，则将等待其完成任务之后删除
  • 代码实现
int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
	if(removing_threads == 0)
		return pool->active_threads;
	
	int remain_threads = pool->active_threads - removing_threads;
	remain_threads = remain_threads > 0 ? remain_threads : 1;

	int i;
	// 循环地取消掉指定数目的线程
	for(i=pool->active_threads-1; i>remain_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;
	}

    return 0;
}

2.7 销毁线程池​

  • 接口文档
原型bool destroy_pool(thread_pool *pool);
功能描述阻塞等待所有任务完成，然后立即销毁整个线程池，释放所有资源和内存
参数pool:将要销毁的线程池
返回值成功返回true，失败返回false
所在头文件thread_pool.h
备注无
  • 代码实现
bool destroy_pool(thread_pool *pool)
{
	pool->shutdown = true;	// 线程池的销毁标记
	pthread_cond_broadcast(&pool->cond);	// 唤醒所有线程

	int i;
	// 等待所有线程退出
	for(i=0; i<pool->active_threads; i++) {
		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]);
		}
	}

	free(pool->task_list);	// 释放资源
	free(pool->tids);		// 清空任务队列
	free(pool);				// 释放任务队列

	return true;
}

2.8 线程入口函数​

  • 用来执行任务的线程函数：
void *routine(void *arg)
{
	thread_pool *pool = (thread_pool *)arg;	//定义线程池结构体指针，拿到线程池的指针
	struct task *p;	//定义任务结构体指针，用来表示拿到的任务

	//不断的拿任务，执行
	while(1) {
		// 访问任务队列前加锁，为防止取消后死锁，注册处理例程 handle
		pthread_cleanup_push(handler, (void *)&pool->lock);	//防止死锁
		pthread_mutex_lock(&pool->lock);	//加锁

		//如果任务队列为空并且线程池没有被销毁，线程进入条件变量等待队列睡眠
		while(pool->waiting_tasks == 0 && !pool->shutdown) {
			//条件变量等待队列
			pthread_cond_wait(&pool->cond, &pool->lock);	
		}

		//如果任务为空，线程池被销毁，则立即释放互斥锁并退出
		if(pool->waiting_tasks == 0 && pool->shutdown) {
			pthread_mutex_unlock(&pool->lock);
			pthread_exit(NULL);
		}

		//取走一个任务
		p = pool->task_list->next;
		pool->task_list->next = p->next;
		pool->waiting_tasks--;	//任务数量减1

		//任务取走，解锁，并弹栈 handle（但不执行它）
		pthread_mutex_unlock(&pool->lock);
		pthread_cleanup_pop(0);

		//执行任务期间拒绝取消请求
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
		(p->task)(p->arg);
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);

		free(p);	//释放资源
	}

	pthread_exit(NULL);
}
  • 注册死锁处理函数
void handler(void *arg)
{
	// 响应取消请求之后自动处理：释放互斥锁
	pthread_mutex_unlock((pthread_mutex_t *)arg);
}

3. 使用示例​

#include <stdio.h>
#include "thread_pool.h"

void *my_task(void *arg)
{
	int n = *(int *)arg;

	printf("[%u][%s] ==> job will be done in %d sec...\n", (unsigned)pthread_self(), __FUNCTION__, n);

	sleep(n);

	printf("[%u][%s] ==> job done!\n", (unsigned)pthread_self(), __FUNCTION__);

	return NULL;

	// pthread_exit(NULL);
}

void *count_time(void *arg)
{
	int i = 0;
	while(1) {
		sleep(1);
		printf("sec: %d\n", ++i);
	}
}

int main(int argc, char *argv[])
{
	pthread_t a;
	pthread_create(&a, NULL, count_time, NULL);

	// 1.初始化一个带有2条线程的线程池
	thread_pool *pool = malloc(sizeof(thread_pool));
	init_pool(pool, 1);

	// 2.投入3个任务
	static int temp1 = 1, temp2 = 2, temp3 = 3, temp4 = 4, temp5 = 5;
	printf("throwing 3 tasks...\n");
	add_task(pool, my_task, (void *)&temp1);
	add_task(pool, my_task, (void *)&temp2);
	add_task(pool, my_task, (void *)&temp3);
  
	// 3.显示当前有多少条线程
	printf("current thread number: %d\n", remove_thread(pool, 0));
	// sleep(9);

	// 4.再投入2个任务
	printf("throwing another 2 tasks...\n");
	add_task(pool, my_task, (void *)&temp4);
	add_task(pool, my_task, (void *)&temp5);

	// 5.增加2条线程
	add_thread(pool, 2);
	sleep(5);

	// 6.删除3条线程
	printf("remove 3 threads from the pool...\n");
	printf("current thread number: %d\n", remove_thread(pool, 3));

	// 7.销毁线程池
	destroy_pool(pool);

	return 0;
}

4. 源码​

想要源码的话，不妨上我的🍍GitHub看看吧！

Loading Comments...
0%