Linux下设计简易线程池
文章目录
- Linux下设计简易线程池
- 1.介绍
- 2.具体实现
- 2.1任务类头文件Task.hpp
- 2.2线程池文件ThreadPool.hpp
- 2.3主函数Main.cc
1.介绍
线程池是一种池化技术，是消费者生产者模型的具体体现。它能够预先创建一批能够被重复使用的线程，而无需创建任何额外的空间。因为减少了线程的创建和开销，所以提高了系统的性能和资源的利用率。
话不多说，在知道具体实现之前，我们先看看一个简易的线程池能够达到的效果：
目的： 运行5个线程完成加减乘除的任务
可以看到，我们运行起来后会有包括主线程在内的6个线程运行起来。在任务方面，我们可以实现每个线程都完成一个任务并返回结果的现象。那么事不宜迟，我们赶快来了解怎么实现一个简易线程池吧！
2.具体实现
首先我们要知道需要了解的知识：
1.线程的使用以及创建(pthread)
2.互斥锁的使用(mutex)
3.条件变量的使用(Condition Variable )
我们的目标：通过一个数组(vector)里存放线程，然后分批完成任务队列(Queue)里的目标，线程任务有则处理，无则等待。期间需要注意使用互斥锁保证原子性，在必要的条件我们需要使用条件变量来唤醒或者等待。在主函数内输入我们的通过输入的数据构建出一个个任务对象，再把任务对象Push进入线程池类中分配线程处理即可。
**代码的结构：**为了使结构清晰，我们打算分三个文件来处理：
1.主函数Main.cc
2.任务类头文件Task.hpp
3.线程池类文件ThreadPool.hpp
2.1任务类头文件Task.hpp
在之前我们已经提过了，我们采用单次加减乘除的任务来模拟日常任务交给线程处理。所以我们的任务Task类就好设计了。
我们类中成员中只用有四个元素即可：第一个数x,第二个数y,操作符ops,结果result ，之后再获取到元素后看成Task对象重写仿函数处理即可。
//模拟任务：实现加减乘除的任务
#pragma once
#include<iostream>
#include<string>
#include<unistd.h>
class Task
{
public:
Task(){}
Task(int x ,int y,char ops)
:_x(x)
,_y(y)
,_ops(ops)
,_result(0)
{}
~Task(){}
void operator()() //仿函数
{
switch(_ops)
{
case '+':
{
_result = _x + _y;
break;
}
case '-':
{
_result = _x - _y;
break;
}
case '*':
{
_result = _x * _y;
break;
}
case '/':
{
if(_y == 0)
{
std::cout<<"zero Error..."<<std::endl;
break;
}
_result = _x / _y;
break;
}
default:
break;
}
usleep(100000);
}
std::string Result() //返回结果
{
return std::to_string(_result);
}
private: //_x + _ y + _ops = _result
int _x;
int _y;
char _ops;
int _result;
};
2.2线程池文件ThreadPool.hpp
在这个线程池类成员中我们打算里面含有：
1.线程存放数组_threads
2.线程个数_num
3.任务队列_tasks
4.互斥锁mtx
5.条件变量_cond
线程池文件代码中有一下这些思想，待会我们展示代码的时候会用到，我们先提前写出来，之后可以跟代码复核：
1.线程池中我们打算采用RAII(资源生命周期随着对象生命周期)思想，线程池的创建(构造)与释放(析构)就包含了互斥锁、条件变量的初始化与销毁。
2.每个线程处理的函数由于是在类内，我们需要定义成static的，否则是不允许调用非静态成员的。既然是类内成员，我们就需要注意隐含有一个this指针，否则可能导致pthread_create创建线程中传参不匹配的问题。
3.每个线程处理的函数很显然是需要上锁的，如果线程队列中已经没有线程可用了，我们就需要使用条件变量等待。
4.线程处理的函数既然是static的，就无法直接使用类内的私有成员函数，就需要定义一批调用函数来间接拿到每个线程的成员如锁、条件变量进行加减锁、唤醒等待。
5.线程池类中当然包括放入任务函数和删除任务的函数，并且需要加锁。
具体对应看代码：
#pragma once
#include<iostream>
#include<string>
#include<vector>
#include<queue>
#include<unistd.h>
#include<pthread.h>
#include "Task.hpp"
const static int N = 5;
template<class T>
class ThreadPool
{
public:
ThreadPool(int num = N) //带参构造
:_num(num)
,_threads(num)
{
pthread_mutex_init(&mtx,nullptr);
pthread_cond_init(&_cond,nullptr);
}
~ThreadPool() //析构函数
{
pthread_mutex_destroy(&mtx);
pthread_cond_destroy(&_cond);
}
//-----------加减锁、唤醒等待的操作接口-----------
void GetLocked() {pthread_mutex_lock(&mtx);}
void GetUnlocked() {pthread_mutex_unlock(&mtx);}
void ThreadWait() {pthread_cond_wait(&_cond,&mtx);}
void ThreadWakeup() {pthread_cond_signal(&_cond);}
bool isEmpty(){return _tasks.empty();}
//--------------------------------------
//放入任务的函数 //需要上锁保证线程安全
T PushTask(const T& t)
{
GetLocked();
_tasks.push(t);
ThreadWakeup();
GetUnlocked();
}
//删除任务
T PopTask()
{
T t = _tasks.front();
_tasks.pop();
return t;
}
//启动函数
void Start()
{
for(int i =0;i<_num;i++)
{
//在类内部一定要注意多一个this指针参数，否则小心传参不正确
pthread_create(&_threads[i],nullptr,threadRoutine,this);
}
}
//--------------------------------------------------
//线程执行函数 （类内需要加上static）否则类内不允许调用内部元素
static void* threadRoutine(void* args)
{
pthread_detach(pthread_self());//先线程自我分离
ThreadPool<T>* tp = static_cast<ThreadPool<T>*>(args); //将当前对象的地址传入this类内中调用
while(true)//线程执行的任务也需要上锁保证原子性
{
//这里需要检测有没有任务
//有则处理，无则等待
//细节也是需要加锁
tp->GetLocked();
while(tp->isEmpty()) //如果线程队列为空则任务需要等待
{
tp->ThreadWait();
}
T t = tp->PopTask(); //从公共区域拿到线程私有区域
tp->GetUnlocked();
t();//执行任务
//输出结果
std::cout<< "thread handler done,result: "<<t.Result()<<std::endl;
}
}
//-----------------------------------------------------------------------
private:
std::vector<pthread_t> _threads; //线程存放数组
int _num; //线程个数
std::queue<T> _tasks; //任务队列
pthread_mutex_t mtx; //互斥锁
pthread_cond_t _cond;//条件变量
};
2.3主函数Main.cc
在主函数中我们为了让RAII思想(资源生命周期随着对象生命周期)得到体现,采用智能指针unique_ptr来让资源得到创建和销毁。
并且我们需要在主函数中获取我们执行加减乘除任务的参数，并把它实例化成任务对象推送到线程池解决。
#include "ThreadPool.hpp"
#include "Task.hpp"
#include<unistd.h>
#include<memory>
int main()
{
std::unique_ptr<ThreadPool<Task>> tp(new ThreadPool<Task>());
tp->Start(); //调用线程池中的创建线程的函数
while(true)
{
int x;
int y;
char ops;
std::cout<<"Please Enter x:";
std::cin >> x;
std::cout<<"Please Enter y:";
std::cin >> y;
std::cout<<"Please Enter ops(+=*/):";
std::cin >> ops;
Task t(x,y,ops);
tp->PushTask(t);
sleep(1);
}
return 0;
}
之后一综合运行就能看到我们一开始看到的结果啦！至此一个简易的线程池就实现完了
