- 一、线程池模型原理分析
- 1. 1多路IO转接和线程池的比较
- 1.2 线程池线程的创建与维护所需要的参数
- 二、线程池描述结构体
- 三、线程池主函数架构
- 四、线程池创建
- 五、向线程池中添加任务
- 六、子线程回调函数
- 6.1 管理线程回调函数
- 6.2 任务线程回调函数
- 七、源码
- 7.1 threadpool.c
- 7.2 threadpool.h
客户端相当于生产者，服务端相当于消费者。
线程池相比较于传统的多线程模型，每次创建线程和销毁线程的开销是可以被节省的。
故放弃每来一个请求就创建一个线程的机制，采用一次性批量产生线程的方式。
所谓线程“池”，这个池是一个虚拟的概念，指的是产生的线程被保存在的可访问地址。
当没有任务时，线程池里的线程都阻塞（pthread_cond_wait() ）在任务队列不为NULL的条件变量上;
当有客户端发来了任务/请求，server需要去唤醒池中的线程：
两个函数：
- pthread_cond_signal()：默认唤醒（阻塞在对应条件变量上的）一个线程。
- pthread_cond_broadcast()：会唤醒所有（阻塞在对应条件变量上的）的线程，即产生惊群效应
被唤醒的线程从任务队列中取出任务并完成处理后，重新回到线程池中等待任务，即等待被唤醒。
1. 1多路IO转接和线程池的比较可以用Unbuntu查看一个线程池的例子：
打开火狐浏览器，并在shell中输入ps aux | grep firefox：
取出进程ID，然后在shell中输入 ps -L [PID]，可以看到虽然我们只开了一个浏览器，但是后台已经一次性启动了许多线程，这里用的就是线程池：
NLWP标识线程数，这里可以看到线程数是38。
-
多路IO转接处理的是客户端怎么去和服务器建立连接和接受请求的问题；
-
而线程池处理的是服务器拿到请求之后，要怎么处理的问题。
1.2 线程池线程的创建与维护所需要的参数故线程池可以和多路IO转接配合使用以提高效能。
在一开始，需要有初步的线程数，即初始值，如：thread_init_num = 38;
但是显然38个线程是不够用的，因为存在访问高峰的情况；当访问高峰来到时，势必要增加线程数，但并不能够无限的增加线程，因为：① 一个进程能够打开的线程数量是有限的 ② 线程虽然和进程共享地址空间，但其栈是要独立的；
故需要一个值来限制最多开启的线程数，如：thread_max_num = 500;
“win系统默认的栈空间是1M(1MiB)大小,而Linux默认下栈空间常见的是8M或10M。
此外，为了判断线程池扩容和瘦身的最佳时机，还需要两个值：
- 记录当前的忙线程数的值：thread_busy_num;
- 记录当前总共的/存活的线程数的值：thread_live_num;
根据这两个值的比例来判判定是否扩容or瘦身；
扩容时需要一个扩容步长：thread_step，即一次扩容多少线程。
既然存在访问高峰就一定存在访问波谷，若服务器仍维护着高峰时期的线程数，那么就会带来不必要的资源开销（尤其是内存开销），故此时需要瘦身，瘦身步长可以和扩容步长一样，也可以单独设置。
关于线程池扩容or瘦身的行为，一个关键的问题是由谁来计算thread_busy_num和thread_live_num的比值，
让server的主进程来做这件事是不合理的，server主进程的主要任务是监听连接，之所以使用线程池的目的就是不让server主进程分心，将数据处理的工作”外包“给线程池，所以应该额外有一个管理者线程，来维护线程池。
二、线程池描述结构体typedef struct {
void *(*function)(void *);
void *arg;
} threadpool_task_t;
struct threadpool_t {
pthread_mutex_t lock;
pthread_mutex_t thread_counter;
pthread_cond_t queue_not_full;
pthread_cond_t queue_not_empty;
pthread_t *threads;
pthread_t adjust_tid;
threadpool_task_t *task_queue;
int min_thr_num;
int max_thr_num;
int live_thr_num;
int busy_thr_num;
int wait_exit_thr_num;
int queue_front;
int queue_rear;
int queue_size;
int queue_max_size;
int shutdown;
};
三、线程池主函数架构实际使用的线程池可以根据业务需求来增加或者减少结构体的相关属性。
int main(void)
{
threadpool_t *thp = threadpool_create(3,100,100);
printf("pool inited");
//int *num = (int *)malloc(sizeof(int)*20);
int num[20], i;
for (i = 0; i < 20; i++) {
num[i] = i;
printf("add task %dn",i);
threadpool_add(thp, process, (void*)&num[i]);
}
sleep(10);
threadpool_destroy(thp);
return 0;
}
四、线程池创建main()函数架构：
- 创建线程池
- 模拟向线程池添加任务… 借助回调处理任务
- 销毁线程池
//threadpool_create(3,100,100);
threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size)
{
int i;
threadpool_t *pool = NULL;
do {
if((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL) {
printf("malloc threadpool fail");
break;
}
pool->min_thr_num = min_thr_num;
pool->max_thr_num = max_thr_num;
pool->busy_thr_num = 0;
pool->live_thr_num = min_thr_num;
pool->wait_exit_thr_num = 0;
pool->queue_size = 0;
pool->queue_max_size = queue_max_size;
pool->queue_front = 0;
pool->queue_rear = 0;
pool->shutdown = false;
pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num);
if (pool->threads == NULL) {
printf("malloc threads fail");
break;
}
memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num);
pool->task_queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size);
if (pool->task_queue == NULL) {
printf("malloc task_queue fail");
break;
}
if (pthread_mutex_init(&(pool->lock), NULL) != 0
|| pthread_mutex_init(&(pool->thread_counter), NULL) != 0
|| pthread_cond_init(&(pool->queue_not_empty), NULL) != 0
|| pthread_cond_init(&(pool->queue_not_full), NULL) != 0)
{
printf("init the lock or cond fail");
break;
}
for (i = 0; i < min_thr_num; i++) {
pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
printf("start thread 0x%x...n", (unsigned int)pool->threads[i]);
}
pthread_create(&(pool->adjust_tid), NULL, adjust_thread, (void *)pool);
return pool;
} while (0);
threadpool_free(pool);
return NULL;
}
五、向线程池中添加任务pthreadpool_create()函数架构:
- 创建线程池结构体指针
- 初始化线程池结构体（N个成员变量）
- 创建N个任务线程
- 创建1个管理者线程
- 失败时，销毁开辟的所有空间
先简要介绍一下条件变量和互斥锁的配合使用方式：
pthread_cond_XXX()函数与互斥锁mutex的配合：
- pthread_cond_wait()
用于阻塞当前线程，等待别的线程使用pthread_cond_signal()或pthread_cond_broadcast来唤醒它 pthread_cond_wait() 必须与pthread_mutex配套使用。pthread_cond_wait()函数一进入wait状态就会自动release mutex（解锁）。当其他线程通过pthread_cond_signal()或pthread_cond_broadcast，把该线程唤醒，使pthread_cond_wait()通过（返回）时，该线程又自动获得该mutex（上锁）。- pthread_cond_signal函数的作用是发送一个信号给另外一个正在处于阻塞等待状态的线程,使其脱离阻塞状态,继续执行.如果没有线程处在阻塞等待状态,pthread_cond_signal也会成功返回。
使用pthread_cond_signal一般不会有“惊群现象”产生，他最多只给一个线程发信号。假如有多个线程正在阻塞等待着这个条件变量的话，那么是根据各等待线程优先级的高低确定哪个线程接收到信号开始继续执行。如果各线程优先级相同，则根据等待时间的长短来确定哪个线程获得信号。但无论如何一个pthread_cond_signal调用最多发信一次。- 但是pthread_cond_signal在多处理器上可能同时唤醒多个线程，当你只能让一个线程处理某个任务时，其它被唤醒的线程就需要继续
wait，而且规范要求pthread_cond_signal至少唤醒一个pthread_cond_wait上的线程，其实有些实现为了简单在单处理器上也会唤醒多个线程。另外，某些应用，如线程池，pthread_cond_broadcast唤醒全部线程，但我们通常只需要一部分线程去做执行任务，所以其它的线程需要继续wait。所以强烈推荐对pthread_cond_wait()使用while循环来做条件判断（下面的任务线程回调函数中就有体现）。
向线程池中添加任务
void *process(void *arg)
{
printf("thread 0x%x working on task %dn ",(unsigned int)pthread_self(),(int)arg);
sleep(1); //模拟 工作
printf("task %d is endn",(int)arg);
return NULL;
}
//调用： threadpool_add(thp, process, (void*)&num[i]);
int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg)
{
pthread_mutex_lock(&(pool->lock));
while ((pool->queue_size == pool->queue_max_size) && (!pool->shutdown)) {
pthread_cond_wait(&(pool->queue_not_full), &(pool->lock));
}
if (pool->shutdown) {
pthread_cond_broadcast(&(pool->queue_not_empty));
pthread_mutex_unlock(&(pool->lock));
return 0;
}
if (pool->task_queue[pool->queue_rear].arg != NULL) {
pool->task_queue[pool->queue_rear].arg = NULL;
}
pool->task_queue[pool->queue_rear].function = function;
pool->task_queue[pool->queue_rear].arg = arg;
pool->queue_rear = (pool->queue_rear + 1) % pool->queue_max_size;
pool->queue_size++;
pthread_cond_signal(&(pool->queue_not_empty));
pthread_mutex_unlock(&(pool->lock));
return 0;
}
六、子线程回调函数 6.1 管理线程回调函数具体的业务可以在void *process(void *arg)函数中自定义。
void *adjust_thread(void *threadpool)
{
int i;
threadpool_t *pool = (threadpool_t *)threadpool;
while (!pool->shutdown) {
sleep(DEFAULT_TIME);
pthread_mutex_lock(&(pool->lock));
int queue_size = pool->queue_size;
int live_thr_num = pool->live_thr_num;
pthread_mutex_unlock(&(pool->lock));
pthread_mutex_lock(&(pool->thread_counter));
int busy_thr_num = pool->busy_thr_num;
pthread_mutex_unlock(&(pool->thread_counter));
if (queue_size >= MIN_WAIT_TASK_NUM && live_thr_num < pool->max_thr_num) {
pthread_mutex_lock(&(pool->lock));
int add = 0;
for (i = 0; i < pool->max_thr_num && add < DEFAULT_THREAD_VARY
&& pool->live_thr_num < pool->max_thr_num; i++) {
if (pool->threads[i] == 0 || !is_thread_alive(pool->threads[i])) {
pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
add++;
pool->live_thr_num++;
}
}
pthread_mutex_unlock(&(pool->lock));
}
if ((busy_thr_num * 2) < live_thr_num && live_thr_num > pool->min_thr_num) {
pthread_mutex_lock(&(pool->lock));
pool->wait_exit_thr_num = DEFAULT_THREAD_VARY;
pthread_mutex_unlock(&(pool->lock));
for (i = 0; i < DEFAULT_THREAD_VARY; i++) {
pthread_cond_signal(&(pool->queue_not_empty));
}
}
}
return NULL;
}
6.2 任务线程回调函数adjust_thread()函数架构：
接收回调参数 void *arg --> pool结构体
循环 DEFAULT_TIME 苏醒一次
对扩容和减容的操作一般没有那么紧迫，不需要让管理线程一直参与cpu的争夺，定时苏醒对线程池进行管理即可。
配合互斥锁的使用，获取【任务数】、【存活线程数】和 【忙线程数】
根据既定算法，使用上述三个变量，计算并判断是否进行扩容或减容操作（即是否应该创建or销毁现称此中指定步长的线程）。
void *threadpool_thread(void *threadpool)
{
threadpool_t *pool = (threadpool_t *)threadpool;
threadpool_task_t task;
while (true) {
pthread_mutex_lock(&(pool->lock));
while ((pool->queue_size == 0) && (!pool->shutdown)) {
printf("thread 0x%x is waitingn", (unsigned int)pthread_self());
pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
if (pool->wait_exit_thr_num > 0) {
pool->wait_exit_thr_num--;
if (pool->live_thr_num > pool->min_thr_num) {
printf("thread 0x%x is exitingn", (unsigned int)pthread_self());
pool->live_thr_num--;
pthread_mutex_unlock(&(pool->lock));
pthread_exit(NULL);
}
}
}
if (pool->shutdown) {
pthread_mutex_unlock(&(pool->lock));
printf("thread 0x%x is exitingn", (unsigned int)pthread_self());
pthread_detach(pthread_self());
pthread_exit(NULL);
}
task.function = pool->task_queue[pool->queue_front].function; //取回调函数
task.arg = pool->task_queue[pool->queue_front].arg; //取回调函数的参数
pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size;
pool->queue_size--;
pthread_cond_broadcast(&(pool->queue_not_full));
pthread_mutex_unlock(&(pool->lock));
pthread_mutex_lock(&(pool->thread_counter));
pool->busy_thr_num++;
pthread_mutex_unlock(&(pool->thread_counter));
printf("thread 0x%x start workingn", (unsigned int)pthread_self());
(*(task.function))(task.arg);
//task.function(task.arg);
printf("thread 0x%x end workingn", (unsigned int)pthread_self());
pthread_mutex_lock(&(pool->thread_counter));
pool->busy_thr_num--;
pthread_mutex_unlock(&(pool->thread_counter));
}
pthread_exit(NULL);
}
七、源码 7.1 threadpool.cthreadpool_thread()函数架构：
接受线程回调参数 void *arg --> pool 结构体
加锁：lock整个线程池结构体
判断条件变量queue_not_empty ：
3.1 若任务队列为空：
则调用pthread_cond_wait使线程自身阻塞wait
阻塞在条件变量的线程被唤醒后判断自身是否需要退出（配合管理线程的减容操作，但存活线程数不应少于最小线程数）
判断是否要销毁线程池，若是，则线程自行退出
从任务队列里获取任务, 是一个出队操作
调用pthread_cond_broadcast()通知阻塞在pool->queue_not_full上的线程可以有新的任务添加进来
任务取出后到临时变量task中之后，立即将线程池的互斥锁lock释放
配合互斥锁pool->thread_counter令线程池中的忙线程数变量+1
执行任务
(*(task.function))(task.arg);处理掉一个任务，配合互斥锁pool->thread_counter令忙状态数线程数-1
#include7.2 threadpool.h#include #include #include #include #include #include #include #include "threadpool.h" #define DEFAULT_TIME 10 #define MIN_WAIT_TASK_NUM 10 #define DEFAULT_THREAD_VARY 10 #define true 1 #define false 0 typedef struct { void *(*function)(void *); void *arg; } threadpool_task_t; struct threadpool_t { pthread_mutex_t lock; pthread_mutex_t thread_counter; pthread_cond_t queue_not_full; pthread_cond_t queue_not_empty; pthread_t *threads; pthread_t adjust_tid; threadpool_task_t *task_queue; int min_thr_num; int max_thr_num; int live_thr_num; int busy_thr_num; int wait_exit_thr_num; int queue_front; int queue_rear; int queue_size; int queue_max_size; int shutdown; }; void *threadpool_thread(void *threadpool); void *adjust_thread(void *threadpool); int is_thread_alive(pthread_t tid); int threadpool_free(threadpool_t *pool); //threadpool_create(3,100,100); threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size) { int i; threadpool_t *pool = NULL; do { if((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL) { printf("malloc threadpool fail"); break; } pool->min_thr_num = min_thr_num; pool->max_thr_num = max_thr_num; pool->busy_thr_num = 0; pool->live_thr_num = min_thr_num; pool->wait_exit_thr_num = 0; pool->queue_size = 0; pool->queue_max_size = queue_max_size; pool->queue_front = 0; pool->queue_rear = 0; pool->shutdown = false; pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num); if (pool->threads == NULL) { printf("malloc threads fail"); break; } memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num); pool->task_queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size); if (pool->task_queue == NULL) { printf("malloc task_queue fail"); break; } if (pthread_mutex_init(&(pool->lock), NULL) != 0 || pthread_mutex_init(&(pool->thread_counter), NULL) != 0 || pthread_cond_init(&(pool->queue_not_empty), NULL) != 0 || pthread_cond_init(&(pool->queue_not_full), NULL) != 0) { printf("init the lock or cond fail"); break; } for (i = 0; i < min_thr_num; i++) { pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool); printf("start thread 0x%x...n", (unsigned int)pool->threads[i]); } pthread_create(&(pool->adjust_tid), NULL, adjust_thread, (void *)pool); return pool; } while (0); threadpool_free(pool); return NULL; } //threadpool_add(thp, process, (void*)&num[i]); int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg) { pthread_mutex_lock(&(pool->lock)); while ((pool->queue_size == pool->queue_max_size) && (!pool->shutdown)) { pthread_cond_wait(&(pool->queue_not_full), &(pool->lock)); } if (pool->shutdown) { pthread_cond_broadcast(&(pool->queue_not_empty)); pthread_mutex_unlock(&(pool->lock)); return 0; } if (pool->task_queue[pool->queue_rear].arg != NULL) { pool->task_queue[pool->queue_rear].arg = NULL; } pool->task_queue[pool->queue_rear].function = function; pool->task_queue[pool->queue_rear].arg = arg; pool->queue_rear = (pool->queue_rear + 1) % pool->queue_max_size; pool->queue_size++; pthread_cond_signal(&(pool->queue_not_empty)); pthread_mutex_unlock(&(pool->lock)); return 0; } void *threadpool_thread(void *threadpool) { threadpool_t *pool = (threadpool_t *)threadpool; threadpool_task_t task; while (true) { pthread_mutex_lock(&(pool->lock)); while ((pool->queue_size == 0) && (!pool->shutdown)) { printf("thread 0x%x is waitingn", (unsigned int)pthread_self()); pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock)); if (pool->wait_exit_thr_num > 0) { pool->wait_exit_thr_num--; if (pool->live_thr_num > pool->min_thr_num) { printf("thread 0x%x is exitingn", (unsigned int)pthread_self()); pool->live_thr_num--; pthread_mutex_unlock(&(pool->lock)); pthread_exit(NULL); } } } if (pool->shutdown) { pthread_mutex_unlock(&(pool->lock)); printf("thread 0x%x is exitingn", (unsigned int)pthread_self()); pthread_detach(pthread_self()); pthread_exit(NULL); } task.function = pool->task_queue[pool->queue_front].function; //取回调函数 task.arg = pool->task_queue[pool->queue_front].arg; //取回调函数的参数 pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size; pool->queue_size--; pthread_cond_broadcast(&(pool->queue_not_full)); pthread_mutex_unlock(&(pool->lock)); pthread_mutex_lock(&(pool->thread_counter)); pool->busy_thr_num++; pthread_mutex_unlock(&(pool->thread_counter)); printf("thread 0x%x start workingn", (unsigned int)pthread_self()); (*(task.function))(task.arg); //task.function(task.arg); printf("thread 0x%x end workingn", (unsigned int)pthread_self()); pthread_mutex_lock(&(pool->thread_counter)); pool->busy_thr_num--; pthread_mutex_unlock(&(pool->thread_counter)); } pthread_exit(NULL); } void *adjust_thread(void *threadpool) { int i; threadpool_t *pool = (threadpool_t *)threadpool; while (!pool->shutdown) { sleep(DEFAULT_TIME); pthread_mutex_lock(&(pool->lock)); int queue_size = pool->queue_size; int live_thr_num = pool->live_thr_num; pthread_mutex_unlock(&(pool->lock)); pthread_mutex_lock(&(pool->thread_counter)); int busy_thr_num = pool->busy_thr_num; pthread_mutex_unlock(&(pool->thread_counter)); if (queue_size >= MIN_WAIT_TASK_NUM && live_thr_num < pool->max_thr_num) { pthread_mutex_lock(&(pool->lock)); int add = 0; for (i = 0; i < pool->max_thr_num && add < DEFAULT_THREAD_VARY && pool->live_thr_num < pool->max_thr_num; i++) { if (pool->threads[i] == 0 || !is_thread_alive(pool->threads[i])) { pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool); add++; pool->live_thr_num++; } } pthread_mutex_unlock(&(pool->lock)); } if ((busy_thr_num * 2) < live_thr_num && live_thr_num > pool->min_thr_num) { pthread_mutex_lock(&(pool->lock)); pool->wait_exit_thr_num = DEFAULT_THREAD_VARY; pthread_mutex_unlock(&(pool->lock)); for (i = 0; i < DEFAULT_THREAD_VARY; i++) { pthread_cond_signal(&(pool->queue_not_empty)); } } } return NULL; } int threadpool_destroy(threadpool_t *pool) { int i; if (pool == NULL) { return -1; } pool->shutdown = true; pthread_join(pool->adjust_tid, NULL); for (i = 0; i < pool->live_thr_num; i++) { pthread_cond_broadcast(&(pool->queue_not_empty)); } for (i = 0; i < pool->live_thr_num; i++) { pthread_join(pool->threads[i], NULL); } threadpool_free(pool); return 0; } int threadpool_free(threadpool_t *pool) { if (pool == NULL) { return -1; } if (pool->task_queue) { free(pool->task_queue); } if (pool->threads) { free(pool->threads); pthread_mutex_lock(&(pool->lock)); pthread_mutex_destroy(&(pool->lock)); pthread_mutex_lock(&(pool->thread_counter)); pthread_mutex_destroy(&(pool->thread_counter)); pthread_cond_destroy(&(pool->queue_not_empty)); pthread_cond_destroy(&(pool->queue_not_full)); } free(pool); pool = NULL; return 0; } int threadpool_all_threadnum(threadpool_t *pool) { int all_threadnum = -1; // 总线程数 pthread_mutex_lock(&(pool->lock)); all_threadnum = pool->live_thr_num; // 存活线程数 pthread_mutex_unlock(&(pool->lock)); return all_threadnum; } int threadpool_busy_threadnum(threadpool_t *pool) { int busy_threadnum = -1; // 忙线程数 pthread_mutex_lock(&(pool->thread_counter)); busy_threadnum = pool->busy_thr_num; pthread_mutex_unlock(&(pool->thread_counter)); return busy_threadnum; } int is_thread_alive(pthread_t tid) { int kill_rc = pthread_kill(tid, 0); //发0号信号，测试线程是否存活 if (kill_rc == ESRCH) { return false; } return true; } #if 1 void *process(void *arg) { printf("thread 0x%x working on task %dn ",(unsigned int)pthread_self(),(int)arg); sleep(1); //模拟 工作 printf("task %d is endn",(int)arg); return NULL; } int main(void) { threadpool_t *thp = threadpool_create(3,100,100); printf("pool inited"); //int *num = (int *)malloc(sizeof(int)*20); int num[20], i; for (i = 0; i < 20; i++) { num[i] = i; printf("add task %dn",i); threadpool_add(thp, process, (void*)&num[i]); } sleep(10); threadpool_destroy(thp); return 0; } #endif
#ifndef __THREADPOOL_H_ #define __THREADPOOL_H_ typedef struct threadpool_t threadpool_t; threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size); int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg); int threadpool_destroy(threadpool_t *pool); int threadpool_all_threadnum(threadpool_t *pool); int threadpool_busy_threadnum(threadpool_t *pool); #endif
