线程池
介绍
线程池： 一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量
线程池的价值：
- 需要大量的线程来完成任务，且完成任务的时间比较短。可同时处理多任务，多请求。
- 有任务可以立即从线程池中调取线程取处理，节省了线程创建的时间
- 有效防止服务端线程过多而导致系统过载的问题
实现
线程池中首先需要有很多个线程，用户可以自己选择创建多少个线程。为了实现线程间的同步与互斥，还需要增加两个变量——互斥量和条件变量。我们还需要一个任务队列，主线程不断往里面塞任务，线程池的线程不断去处理。需要注意的是：这里的任务队列可以为空，但不能满，所以任务队列的容量不限定（实际场景中，任务队列容量不够就需要考虑换一台更大的服务器）
线程池的四个成员变量：
- 一个队列： 存放任务
- 线程池中线程数： 记录线程池中创建的线程数
- 互斥量： 一个互斥锁
- 条件变量： 两个条件变量
线程池：首先需要创建几个线程，还有一个任务队列，当任务队列有任务的时候就唤醒一个正在等待的线程，让线程去执行任务，线程池中的线程执行完任务不会销毁，大大减少的cpu的消耗。
需要两个条件变量和一个互斥锁，这个互斥锁用来锁住任务队列，因为任务队列是公共资源，其次还需要两个条件变量，一个条件变量用来阻塞取任务的线程，当队列中有任务的时候，直接取任务，然后解锁，当任务队列中没有任务的时候，解锁等待条件，条件满足抢锁，取任务，解锁。另一个条件变量用来阻塞添加者进程，当任务队列满了，会让添加者进程等待，当有线程取走一个任务的时候，会唤醒添加者进程。
版本一
任务函数
这里的任务函数采用的时回调函数的方式，提高了代码的通用性，可以根据自己的需求改写任务函数
//任务回调函数
void taskRun(void *arg)
{
PoolTask *task = (PoolTask*)arg;
int num = task->tasknum;
printf("task %d is runing %lu\n",num,pthread_self());
sleep(1);
printf("task %d is done %lu\n",num,pthread_self());
}
线程池的主要代码框架
class ThreadPool
{
public:
//构造函数，初始化线程池
ThreadPool(int thrnum, int maxtasknum)
{
}
static void* thrRun(void* arg)
{
}
//析构函数，摧毁线程池
~ThreadPool()
{
}
public:
//添加任务到线程池
void addtask(){}；
private:
//任务队列相关的参数
int max_job_num;//最大任务个数
int job_num;//实际任务个数
PoolTask *tasks;//任务队列数组
int job_push;//入队位置
int job_pop;// 出队位置
//线程相关参数
int thr_num;//线程池内线程个数
pthread_t *threads;//线程池内线程数组
int shutdown;//是否关闭线程池
pthread_mutex_t pool_lock;//线程池的锁
pthread_cond_t empty_task;//任务队列为空的条件
pthread_cond_t not_empty_task;//任务队列不为空的条件
};
放任务： 主线程无脑往任务队列中塞任务，塞任务之前进行加锁，塞完任务解锁，如果任务队列已经满了，等待线程取任务，然后唤醒在条件变量下等待的队列；放入了任务就给线程发送信号，唤醒线程来取
取任务： 线程池中的线程从任务队列中取任务，需要对任务队列上锁，因为对公共资源的操作都需要上锁，如果没有任务就阻塞，等待放任务唤醒；如果取完了一个任务，就唤醒添加任务
这就是两个条件变量和一个互斥锁的用法
//添加任务到线程池
void addtask()
{
pthread_mutex_lock(&(this->pool_lock));
//实际任务总数大于最大任务个数则阻塞等待(等待任务被处理)
while(this->max_job_num <= this->job_num)
{
pthread_cond_wait(&(this->empty_task),&(this->pool_lock));
}
int taskpos = (this->job_push++)%this->max_job_num;
this->tasks[taskpos].tasknum = beginnum++;
this->tasks[taskpos].arg = (void*)&this->tasks[taskpos];
this->tasks[taskpos].task_func = taskRun;
this->job_num++;
pthread_mutex_unlock(&(this->pool_lock));
pthread_cond_signal(&(this->not_empty_task));//通知包身工
}
//取任务
static void* thrRun(void* arg)
{
ThreadPool *pool = (ThreadPool*)arg;
int taskpos = 0;//任务位置
PoolTask *task = new PoolTask();
while(1)
{
//获取任务，先要尝试加锁
pthread_mutex_lock(&pool->pool_lock);
//无任务并且线程池不是要摧毁
while(pool->job_num <= 0 && !pool->shutdown )
{
//如果没有任务，线程会阻塞
pthread_cond_wait(&pool->not_empty_task,&pool->pool_lock);
}
if(pool->job_num)
{
//有任务需要处理
taskpos = (pool->job_pop++)%pool->max_job_num;
//printf("task out %d...tasknum===%d tid=%lu\n",taskpos,thrPool->tasks[taskpos].tasknum,pthread_self());
//为什么要拷贝？避免任务被修改，生产者会添加任务
memcpy(task,&pool->tasks[taskpos],sizeof(PoolTask));
task->arg = task;
pool->job_num--;
//task = &thrPool->tasks[taskpos];
pthread_cond_signal(&pool->empty_task);//通知生产者
}
if(pool->shutdown)
{
//代表要摧毁线程池，此时线程退出即可
//pthread_detach(pthread_self());//临死前分家
pthread_mutex_unlock(&pool->pool_lock);
delete(task);
pthread_exit(NULL);
}
//释放锁
pthread_mutex_unlock(&pool->pool_lock);
task->task_func(task->arg);//执行回调函数
}
}
整体代码：
#include<iostream>
#include<string.h>
#include<pthread.h>
#include<sys/types.h>
#include<stdio.h>
#include<unistd.h>
using namespace std;
int beginnum = 1;
class PoolTask
{
public:
int tasknum;//模拟任务编号
void *arg;//回调函数参数
void (*task_func)(void *arg);//任务的回调函数
};
//任务回调函数
void taskRun(void *arg)
{
PoolTask *task = (PoolTask*)arg;
int num = task->tasknum;
printf("task %d is runing %lu\n",num,pthread_self());
sleep(1);
printf("task %d is done %lu\n",num,pthread_self());
}
class ThreadPool
{
public:
//构造函数，初始化线程池
ThreadPool(int thrnum, int maxtasknum)
{
this->thr_num = thrnum;
this->max_job_num = maxtasknum;
this->shutdown = 0;//是否摧毁线程池，1代表摧毁
this->job_push = 0;//任务队列添加的位置
this->job_pop = 0;//任务队列出队的位置
this->job_num = 0;//初始化的任务个数为0
//申请最大的任务队列
this->tasks = new PoolTask[thrnum];
//初始化锁和条件变量
pthread_mutex_init(&(this->pool_lock),NULL);
pthread_cond_init(&(this->empty_task),NULL);
pthread_cond_init(&(this->not_empty_task),NULL);
int i = 0;
this->threads = (pthread_t *)malloc(sizeof(pthread_t)*thrnum);//申请n个线程id的空间
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
for(i = 0;i < thrnum;i++)
{
pthread_create(&(this->threads[i]),&attr,thrRun,this);//创建多个线程
}
}
static void* thrRun(void* arg)
{
ThreadPool *pool = (ThreadPool*)arg;
int taskpos = 0;//任务位置
PoolTask *task = new PoolTask();
while(1)
{
//获取任务，先要尝试加锁
pthread_mutex_lock(&pool->pool_lock);
//无任务并且线程池不是要摧毁
while(pool->job_num <= 0 && !pool->shutdown )
{
//如果没有任务，线程会阻塞
pthread_cond_wait(&pool->not_empty_task,&pool->pool_lock);
}
if(pool->job_num)
{
//有任务需要处理
taskpos = (pool->job_pop++)%pool->max_job_num;
//printf("task out %d...tasknum===%d tid=%lu\n",taskpos,thrPool->tasks[taskpos].tasknum,pthread_self());
//为什么要拷贝？避免任务被修改，生产者会添加任务
memcpy(task,&pool->tasks[taskpos],sizeof(PoolTask));
task->arg = task;
pool->job_num--;
//task = &thrPool->tasks[taskpos];
pthread_cond_signal(&pool->empty_task);//通知生产者
}
if(pool->shutdown)
{
//代表要摧毁线程池，此时线程退出即可
//pthread_detach(pthread_self());//临死前分家
pthread_mutex_unlock(&pool->pool_lock);
delete(task);
pthread_exit(NULL);
}
//释放锁
pthread_mutex_unlock(&pool->pool_lock);
task->task_func(task->arg);//执行回调函数
}
}
//析构函数，摧毁线程池
~ThreadPool()
{
this->shutdown = 1;//关闭线程池
pthread_cond_broadcast(&(this->not_empty_task));//诱杀
int i = 0;
for(i = 0; i<this->thr_num ; i++)
{
pthread_join(this->threads[i],NULL);
}
pthread_cond_destroy(&(this->not_empty_task));
pthread_cond_destroy(&(this->empty_task));
pthread_mutex_destroy(&(this->pool_lock));
delete []tasks;
tasks = NULL;
free(this->threads);
}
public:
//添加任务到线程池
void addtask()
{
pthread_mutex_lock(&(this->pool_lock));
cout << "当前任务队列中任务的个数是： " <<this-> job_num <<endl;
//实际任务总数大于最大任务个数则阻塞等待(等待任务被处理)
while(this->max_job_num <= this->job_num)
{
pthread_cond_wait(&(this->empty_task),&(this->pool_lock));
}
int taskpos = (this->job_push++)%this->max_job_num;
this->tasks[taskpos].tasknum = beginnum++;
this->tasks[taskpos].arg = (void*)&this->tasks[taskpos];
this->tasks[taskpos].task_func = taskRun;
this->job_num++;
pthread_mutex_unlock(&(this->pool_lock));
pthread_cond_signal(&(this->not_empty_task));//通知包身工
}
private:
//任务队列相关的参数
int max_job_num;//最大任务个数
int job_num;//实际任务个数
PoolTask *tasks;//任务队列数组
int job_push;//入队位置
int job_pop;// 出队位置
//线程相关参数
int thr_num;//线程池内线程个数
pthread_t *threads;//线程池内线程数组
int shutdown;//是否关闭线程池
pthread_mutex_t pool_lock;//线程池的锁
pthread_cond_t empty_task;//任务队列为空的条件
pthread_cond_t not_empty_task;//任务队列不为空的条件
};
int main()
{
ThreadPool *m = new ThreadPool(3,20);
int j = 0;
for(j=0;j<20;j++)
{
m->addtask();
}
sleep(20);
delete m;
m = NULL;
system("pause");
return EXIT_SUCCESS;
}
运行结果如下：
可以看到线程最多处理三个任务，而任务队列中最多可以存在20个任务，当线程取走了任务之后，唤醒生产者继续添加任务。
版本二
首先封装一个任务：
class Task
{
public:
Task(int a = 0, int b = 0)
:_a(a)
,_b(b)
{}
void Run()
{
//执行的任务可以自己编写
}
private:
int _a;
int _b;
};
线程池的主要代码框架（唤醒和等待操作都已经封装好）：
#define DEFAULT_MAX_PTHREAD 5
class ThreadPool
{
public:
ThreadPool(int max_pthread = DEFAULT_MAX_PTHREAD)
:_max_thread(max_pthread)
{}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
public:
void LockQueue()
{
pthread_mutex_lock(&_mutex);
}
void UnlockQueue()
{
pthread_mutex_unlock(&_mutex);
}
void ThreadWait()
{
pthread_cond_wait(&_cond, &_mutex);
}
void WakeUpThread()
{
pthread_cond_signal(&_cond);
//pthread_cond_broadcast(&_cond);
}
bool IsEmpty()
{
return _q.empty();
}
private:
queue<Task*> _q;
int _max_thread;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
};
创建多个线程
创建多个线程可以用一个循环进行创建。需要注意的是，创建一个线程还需要提供一个线程启动后要执行的函数，这个启动函数只能有一个参数。如果把这个函数设置为成员函数，那么这个函数的第一个参数默认是this指针，这样显然是不可行的，所以这里我们考虑把这个启动函数设置为静态的。但是设置为静态的成员函数又会面临一个问题：如何调用其他成员函数和成员变量？ 所以这里我们考虑创建线程的时候，把this指针传过去，让启动函数的arg 参数去接收即可
static void* Runtine(void* arg)
{
pthread_detach(pthread_self());
ThreadPool* this_p = (ThreadPool*)arg;
while (1){
this_p->LockQueue();
while (this_p->IsEmpty()){
this_p->ThreadWait();
}
Task* t;
this_p->Get(t);
this_p->UnlockQueue();
// 解锁后处理任务
t->Run();
delete t;
}
}
void ThreadPoolInit()
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
pthread_t t[_max_thread];
for(int i = 0; i < _max_thread; ++i)
{
pthread_create(t + i, nullptr, Runtine, this);
}
}
注意： 线程创建后，执行启动函数，在这个函数中，线程会去任务队列中取任务并处理，取任务前需要进行加锁的操作（如果队列为空需要挂起等待），取完任务然后进行解锁，然后处理任务，让其它线程去任务队列中取任务
放任务： 主线程无脑往任务队列中塞任务，塞任务之前进行加锁，塞完任务解锁，然后唤醒在条件变量下等待的队列
取任务： 线程池中的线程从任务队列中取任务，这里不需要加锁，因为这个动作在启动函数中加锁的那一段区间中被调用的，其实已经上锁了
// 放任务
void Put(Task* data)
{
LockQueue();
_q.push(data);
UnlockQueue();
WakeUpThread();
}
// 取任务
void Get(Task*& data)
{
data = _q.front();
_q.pop();
}
这两个版本都可以实现简易的线程池，下面线程池版本的服务器主要是用版本二来实现，因为版本一要修改的内容有点多，小伙伴们可以自己修改一下
线程池版本服务器
多线程版本效果看起来还不错，但是来一个连接就创建一个线程，断开一个连接就释放一个线程，这样频繁地创建和释放线程资源，对OS来说是一种负担，同时也带来资源的浪费，如果我们使用线程池，把每一个客户端连接封装成一个任务，让线程池去处理，这样就不需要频繁地创建和销毁消除，效率也能提升很多。
线程池采用版本二，代码如下：
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#include <unistd.h>
#include "Task.hpp"
#define DEFAULT_MAX_PTHREAD 5
class ThreadPool
{
public:
ThreadPool(int max_pthread = DEFAULT_MAX_PTHREAD)
:_max_thread(max_pthread)
{}
static void* Runtine(void* arg)
{
pthread_detach(pthread_self());
ThreadPool* this_p = (ThreadPool*)arg;
while (1){
this_p->LockQueue();
while (this_p->IsEmpty()){
this_p->ThreadWait();
}
Task* t;
this_p->Get(t);
this_p->UnlockQueue();
// 解锁后处理任务
t->Run();
delete t;
}
}
void ThreadPoolInit()
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
pthread_t t[_max_thread];
for(int i = 0; i < _max_thread; ++i)
{
pthread_create(t + i, nullptr, Runtine, this);
}
}
void Put(Task* data)
{
LockQueue();
_q.push(data);
UnlockQueue();
WakeUpThread();
}
void Get(Task*& data)
{
data = _q.front();
_q.pop();
}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
public:
void LockQueue()
{
pthread_mutex_lock(&_mutex);
}
void UnlockQueue()
{
pthread_mutex_unlock(&_mutex);
}
void ThreadWait()
{
pthread_cond_wait(&_cond, &_mutex);
}
void WakeUpThread()
{
pthread_cond_signal(&_cond);
//pthread_cond_broadcast(&_cond);
}
bool IsEmpty()
{
return _q.empty();
}
private:
std::queue<Task*> _q;
int _max_thread;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
};
这里我们单独写一个头文件——Task.hpp，其中有任务类，任务类里面有三个成员变量，也就是端口号，IP和套接字，其中有一个成员方法——Run，里面封装了一个Service函数，也就是前面写的，把它放在Task.hpp这个头文件下，线程池里面的线程执行run函数即可，头文件内容如下：
#pragma once
#include <iostream>
#include <unistd.h>
static void Service(std::string ip, int port, int sock)
{
while (1){
char buf[256];
ssize_t size = read(sock, buf, sizeof(buf)-1);
if (size > 0){
// 正常读取size字节的数据
buf[size] = 0;
std::cout << "[" << ip << "]:[" << port << "]# "<< buf << std::endl;
std::string msg = "server get!-> ";
msg += buf;
write(sock, msg.c_str(), msg.size());
}
else if (size == 0){
// 对端关闭
std::cout << "[" << ip << "]:[" << port << "]# close" << std::endl;
break;
}
else{
// 出错
std::cerr << sock << "read error" << std::endl;
break;
}
}
close(sock);
std::cout << "service done" << std::endl;
}
struct Task
{
int _port;
std::string _ip;
int _sock;
Task(int port, std::string ip, int sock)
:_port(port)
,_ip(ip)
,_sock(sock)
{}
void Run()
{
Service(_ip, _port, _sock);
}
};
服务器类的核心代码如下：
void loop()
{
struct sockaddr_in peer;// 获取远端端口号和ip信息
socklen_t len = sizeof(peer);
_tp = new ThreadPool(THREAD_NUM);
_tp->ThreadPoolInit();
while (1){
// 获取链接
// sock 是进行通信的一个套接字 _listen_sock 是进行监听获取链接的一个套接字
int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
if (sock < 0){
std::cout << "accept fail, continue accept" << std::endl;
continue;
}
int peerPort = ntohs(peer.sin_port);
std::string peerIp = inet_ntoa(peer.sin_addr);
std::cout << "get a new link, [" << peerIp << "]:[" << peerPort << "]"<< std::endl;
Task* task = new Task(peerPort, peerIp, sock);
_tp->Put(task);
}
}
注意几点变化：
- 服务器类增加一个线程池成员变量，初始化函数里面增加线程池创建（在堆上申请）
- 析构函数增加释放线程池资源一步
- loop函数中只需要封装任务，并把任务丢进线程池中即可