线程池

线程池是一种多线程的处理形式，处理过程中将任务添加到任务队列中，然后创建线程后启动这些任务

注意：

1，任务队列中刚开始没有任何任务，是一个具有头结点的空链队列。

2，使用互斥锁来保护这个队列。

3，使用条件变量来代表任务队列中的任务个数的变化——将来如果主线程往队列中投放任务，那么可以通过条件变量来唤醒那些睡着了的线程。

4，通过一个公共开关——shutdown，来控制线程退出，进而销毁整个线程池。

1.API接口设计

1、线程池相关的结构体

一个任务的结构体

任务节点，包含需要执行的函数及其参数，通过链表连成一个任务队列

struct task 	//任务节点
{
	void *(*task)(void *arg);	//任务函数
	void *arg;					//参数
	struct task* next;			//后继指针	
};
//任务实例最终形成一条单向链表
1
2
3
4
5
6
7

线程池实例，包含一个线程池的所有信息

struct thread_pool	//线程池
{
	pthread_mutex_t lock; 			// 互斥锁，保护任务队列
	pthread_cond_t cond; 			// 条件变量，同步所有线程
	struct task *task_list; 		// 任务链队列指针
	pthread_t *tids; 				// 线程 ID 存放位置
	bool shutdown; 					// 线程池销毁标记
	unsigned int waiting_tasks; 	// 任务链队列中等待的任务个数
	unsigned int active_threads;	// 当前活跃线程个数
}
//活跃线程个数可修改，但至少有 1 条活跃线程
1
2
3
4
5
6
7
8
9
10
11

2.线程池初始化：init_pool()

功能描述
	创建一个新的线程池，包含threads_number个活跃线程pool:线程池指针
原型
	bool init_pool(thread_pool *pool,unsigned int threads_number);
参数
	threads_number:初始活跃线程个数(大于等于1)
返回值
	成功返回true，失败返回false
备注
	线程池最少线程个数为1
1
2
3
4
5
6
7
8
9
10

3.增加活跃线程：add_thread( )

功能描述
	增加线程池中活跃线程的个数
原型
	int add_thread(thread_pool *pool,unsigned int additional_threads);
参数
	pool: 	需要增加线程的线程池指针
	additional_threads:	新增线程个数
返回值
	>0:	实际新增线程个数
	-1:失败
1
2
3
4
5
6
7
8
9
10

4.删除活跃线程：remove_thread()

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

1
2
3
4
5
6
7
8
9
10
11
12
13

5.销毁线程池：destroy_pool()

功能描述
	阻塞等待所有任务完成，然后立即销毁整个线程池，释放所有资源和内存
原型
	bool destroy_pool(thread_pool *pool);
参数
	pool:将要销毁的线程池
返回值
	成功返回true，失败返回false
1
2
3
4
5
6
7
8

thread_pool.h

#ifndef _THREAD_POOL_H_
#define _THREAD_POOL_H_

#include <stdio.h>
#include <stdbool.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>

#include <errno.h>
#include <pthread.h>

#define MAX_WAITING_TASKS	1000
#define MAX_ACTIVE_THREADS	20

//任务结构体
struct task
{
	//任务接口
	void *(*task)(void *arg);
	void *arg;
	//任务链表
	struct task *next;
};

//线程池的组成部分
typedef struct thread_pool
{
	pthread_mutex_t lock;		//互斥锁
	pthread_cond_t  cond;		//条件变量
	struct task *task_list;		//任务结构体

	pthread_t *tids;			//存放tid

	unsigned waiting_tasks;		//等待任务数
	unsigned active_threads;	//活跃线程数

	bool shutdown;				//关闭标志
}thread_pool;


//初始化线程池
bool
init_pool(thread_pool *pool,
          unsigned int threads_number);

//增加任务
bool
add_task(thread_pool *pool,
         void *(*task)(void *arg),
         void *arg);
//增加线程
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

thread_pool.c

void handler(void *arg)
{

	pthread_mutex_unlock((pthread_mutex_t *)arg);
}

//线程函数
void *routine(void *arg)
{
	thread_pool *pool = (thread_pool *)arg;
	struct task *p;

	while(1)
	{
		//防止死锁
		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 == true)
		{
			pthread_mutex_unlock(&pool->lock);
			pthread_exit(NULL);
		}

		//取节点，删除节点
		p = pool->task_list->next;
		pool->task_list->next = p->next;
		//任务数-1
		pool->waiting_tasks--;

		//解锁
		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);
}

//初始化线程池
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);

	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;
}

//增加任务
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;
	//任务数+1
	pool->waiting_tasks++;


	pthread_mutex_unlock(&pool->lock);
	pthread_cond_signal(&pool->cond);

	return true;
}

//增加线程
int add_thread(thread_pool *pool, unsigned 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;
}

//删除线程
int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
	//如果参数为0，则可以查看当前活跃线程数
	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;
	}
}

//销毁线程池
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;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228

参考：

线程池