1. 概念
安检
银行柜台
- 为什么使用线程池？
频繁创建和销毁线程浪费CPU资源 - 线程是什么？
一堆线程放在一个池子里统一管理
2. 构成
线程池
2.1 任务队列job_queue
- 作用
存放待处理的任务 - 成员
| No. | 构成 | 接口 |
|---|---|---|
| 1 | 处理函数 | void *(*)(void*) |
| 2 | 参数 | void *arg |
| 3 | 队列指针 | struct job_queue* pnext |
2.2 工作线程worker
- 作用
处理任务
2.3 线程池thread_pool
作用
管理多个线程并提供任务队列的接口成员
| No. | 构成 | 接口 |
|---|---|---|
| 1 | 初始化 | threadpool_init() |
| 2 | 销毁 | threadpool_destroy() |
| 3 | 添加任务 | threadpool_add_job() |
3. 流程
使用流程
- 初始化线程池
- 向线程池添加任务
- 销毁线程池
线程池初始化
- 初始化任务队列和工作线程组
- 将等候在条件变量（任务队列上有任务）上的一个线程唤醒并从该任务队列中取出第一个任务给该线程执行
- 等待任务队列中所有任务执行完毕
4. 实例
- Linux C语言实现
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <assert.h>
struct job_queue{
void* (*func)(void* arg);
void* arg;
struct job_queue* pnext;
};
struct threadpool{
struct job_queue* phead;
struct job_queue* ptail;
pthread_t* pworks;
size_t nthread;
pthread_mutex_t mutex;
pthread_cond_t cond;
int destroy;
};
void* threadpool_routine(void* arg){
assert(NULL != arg);
struct threadpool* pthpool = (struct threadpool*)arg;
for(;;){
struct job_queue* pjob = NULL;
pthread_mutex_lock(&(pthpool->mutex));
while(pthpool->phead == NULL && 0 == pthpool->destroy){
pthread_cond_wait(&(pthpool->cond),&(pthpool->mutex));
}
//
if(1 == pthpool->destroy){
printf("%lu exit\n",pthread_self());
pthread_mutex_unlock(&(pthpool->mutex));
pthread_exit(NULL);
}
pjob = pthpool->phead;
pthpool->phead = pjob->pnext;
pthread_mutex_unlock(&(pthpool->mutex));
pjob->func(pjob->arg);
free(pjob);
}
}
struct threadpool* threadpool_init(size_t nthread){
struct threadpool* pthpool = malloc(sizeof(struct threadpool));
pthread_mutex_init(&(pthpool->mutex),NULL);
pthread_cond_init(&(pthpool->cond),NULL);
pthpool->phead = NULL;
pthpool->ptail = NULL;
pthpool->nthread = nthread;
pthpool->destroy = 0;
// 初始化工作线程组
pthpool->pworks = malloc(nthread * sizeof(pthread_t));
int i;
for(i=0;i<nthread;i++){
pthread_create(&(pthpool->pworks[i]),NULL,threadpool_routine,pthpool);
}
return pthpool;
}
void threadpool_destroy(struct threadpool* pthpool){
assert(NULL != pthpool);
while(NULL != pthpool->phead){
struct job_queue* pnext = pthpool->phead->pnext;
free(pthpool->phead);
pthpool->phead = pnext;
}
pthpool->destroy = 1;
pthread_cond_broadcast(&(pthpool->cond));
int i;
for(i=0;i<pthpool->nthread;i++){
pthread_join(pthpool->pworks[i],NULL);
}
free(pthpool->pworks);
free(pthpool);
pthread_mutex_destroy(&(pthpool->mutex));
pthread_cond_destroy(&(pthpool->cond));
pthpool = NULL;
}
void threadpool_add_job(struct threadpool* pthpool,void*(*func)(void*),void* arg){
assert(NULL != pthpool);
struct job_queue* jq = malloc(sizeof(struct job_queue));
jq->func = func;
jq->arg = arg;
jq->pnext = NULL;
printf("add job %d\n",(int) arg);
pthread_mutex_lock(&(pthpool->mutex));
if(pthpool->phead == NULL){
pthpool->phead = jq;
pthpool->ptail = jq;
}else{
pthpool->ptail->pnext = jq;
pthpool->ptail = jq;
}
pthread_mutex_unlock(&(pthpool->mutex));
pthread_cond_signal(&(pthpool->cond));
}
void* test(void* arg){
printf("%lu do job %d\n",pthread_self(),(int)arg);
sleep(1);
}
int main(int argc,char* argv[]){
if(3 != argc){
printf("usage:%s\n <#nthread> <#njob>", argv[0]);
return 1;
}
int nthread = atoi(argv[1]);
int njob = atoi(argv[2]);
// 1. 初始化线程池
struct threadpool* pool = threadpool_init(nthread);
// 2. 向线程池添加任务
int i;
for(i=0;i<njob;i++){
threadpool_add_job(pool,test,(void*)i);
}
pause();
// 3. 销毁线程池
threadpool_destroy(pool);
}
注意：任务队列初始化在工作线程组前。
- Linux C++98实现
#include <pthread.h>
#include <vector>
#include <queue>
#include <unistd.h>
#include <cstdio>
#include <cstdlib>
using namespace std;
class ThreadPool{
typedef void (*func_t)(int);
public:
ThreadPool(size_t count):destroy(false){
// 初始化互斥锁和条件变量
pthread_mutex_init(&mutex,NULL);
pthread_cond_init(&cond,NULL);
// 初始化线程组
for(int i=0;i<count;i++){
pthread_t tid;
pthread_create(&tid,NULL,reinterpret_cast<void*(*)(void*)>(&ThreadPool::Route),this);
threads.push_back(tid);
}
}
~ThreadPool(){
// 通知线程退出
destroy = true;
pthread_cond_broadcast(&cond);
for(vector<pthread_t>::iterator it = threads.begin();it != threads.end();it++){
pthread_join(*it,NULL);
}
// 销毁互斥锁和条件变量
pthread_mutex_destroy(&mutex);
pthread_cond_destroy(&cond);
}
void AddJob(func_t func,int arg){
pthread_mutex_lock(&mutex);
tasks.push(func);
args.push(arg);
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
}
private:
static void Route(ThreadPool* pPool){
for(;;){
pthread_mutex_lock(&(pPool->mutex));
// 如果没有任务等待
while(pPool->tasks.empty() && !pPool->destroy){
pthread_cond_wait(&(pPool->cond),&(pPool->mutex));
}
// 线程退出
if(pPool->destroy){
pthread_mutex_unlock(&(pPool->mutex));
break;
}
// 获取任务
func_t task = pPool->tasks.front();
pPool->tasks.pop();
int arg = pPool->args.front();
pPool->args.pop();
pthread_mutex_unlock(&(pPool->mutex));
// 执行任务
task(arg);
}
}
private:
vector<pthread_t> threads; ///< 线程组
queue<func_t> tasks; ///< 任务队列
queue<int> args; ///< 参数队列
pthread_mutex_t mutex;
pthread_cond_t cond;
bool destroy; ///< 线程池销毁标志
};
void test(int arg){
printf("%lu do job %d\n",pthread_self(),arg);
sleep(1);
}
int main(int argc,char** argv){
if(3 != argc){
printf("usage:%s\n <#nthread> <#njob>", argv[0]);
return 1;
}
// 1. 初始化线程池
int nthread = atoi(argv[1]);
int njob = atoi(argv[2]);
ThreadPool pool(nthread);
// 2. 向线程池添加任务
for(int i=0;i<njob;i++){
pool.AddJob(test,i);
}
pause();
return 0;
}
- C++11实现
#include <thread>
#include <vector>
#include <queue>
#include <cstdio>
#include <cstdlib>
#include <condition_variable>
#include <mutex>
#include <functional>
#include <chrono>
#include <iostream>
using namespace std;
class ThreadPool{
typedef function<void(int)> func_t;
public:
ThreadPool(size_t count):destroy(false){
// 初始化线程组
for(int i=0;i<count;i++){
threads.emplace_back(&ThreadPool::Route,this);
}
}
~ThreadPool(){
// 通知线程退出
destroy = true;
cond.notify_all();
for(auto& th:threads){
th.join();
}
}
void AddJob(func_t func,int arg){
mtx.lock();
tasks.push(func);
args.push(arg);
cond.notify_one();
mtx.unlock();
}
private:
static void Route(ThreadPool* pPool){
for(;;){
unique_lock<mutex> lck(pPool->mtx);
// 如果没有任务等待
while(pPool->tasks.empty() && !pPool->destroy){
pPool->cond.wait(lck);
}
// 线程退出
if(pPool->destroy){
break;
}
// 获取任务
func_t task = pPool->tasks.front();
pPool->tasks.pop();
int arg = pPool->args.front();
pPool->args.pop();
lck.unlock();
// 执行任务
task(arg);
}
}
private:
vector<thread> threads; ///< 线程组
queue<func_t> tasks; ///< 任务队列
queue<int> args; ///< 参数队列
mutex mtx;
condition_variable cond;
bool destroy; ///< 线程池销毁标志
};
void test(int arg){
cout << this_thread::get_id() << " do job " << arg << endl;
this_thread::sleep_for(chrono::seconds(1));
}
int main(int argc,char** argv){
if(3 != argc){
printf("usage:%s\n <#nthread> <#njob>", argv[0]);
return 1;
}
// 1. 初始化线程池
int nthread = atoi(argv[1]);
int njob = atoi(argv[2]);
ThreadPool pool(nthread);
// 2. 向线程池添加任务
for(int i=0;i<njob;i++){
pool.AddJob(test,i);
}
this_thread::sleep_for(chrono::seconds(10));
return 0;
}
C++11 lambda表达式实现
#include <iostream>
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <functional>
using namespace std;
class ThreadPool{
vector<thread> threads;
queue<function<void()>> tasks;
condition_variable cond;
mutex m;
bool exited = false;
public:
ThreadPool(int num){
auto routine = [this](){
while(true){
unique_lock<mutex> lock(m);
while(tasks.empty() && !exited) cond.wait(lock);
if(exited) break;
auto func = tasks.front();
tasks.pop();
lock.unlock();
func();
}
};
for(int i=0;i<num;++i){
// thread t(routine);
// threads.emplace_back(move(t));
// threads.emplace_back(thread(routine));
threads.emplace_back(thread{routine});
}
}
~ThreadPool(){
exited = true;
cond.notify_all();
for(auto& thread:threads){
thread.join();
}
}
ThreadPool(const ThreadPool&) = default;
ThreadPool& operator=(const ThreadPool&) = default;
void AddTask(function<void()> func){
lock_guard<mutex> guard(m);
tasks.push(func);
cond.notify_one();
// signal
}
};
int main(){
mutex m;
auto test = [&m](){
this_thread::sleep_for(2s);
lock_guard<mutex> guard(m);
cout << this_thread::get_id() << " do something..." << endl;
};
ThreadPool pool(5);
for(int i=0;i<15;++i){
pool.AddTask(test);
}
this_thread::sleep_for(10s);
}
