Linux中线程池的制作
一.介绍
1.1概念
一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。
线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。
可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets
等的数量。
1.2应用场景
- 需要大量的线程来完成任务，且完成任务的时间比较短。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。短时间内产生大量线程可能使内存到达极限，出现错误。
二.线程池的实现
设计一个线程池的类，用队列来实现。指定线程的数量，并提供pop与push接口。
外部线程可通过push接口将任务放入队列中，线程池中多个线程去执行pop的任务。
代码示例：
基本框架：
#define pthnums 3
template<class T>
class threadpool
{
public:
threadpool(int nums=pthnums)
{
pthread_num=nums;
pthread_mutex_init(&mutex_,nullptr);
pthread_cond_init(&cond_,nullptr);
is_start=true;
}
~threadpool()
{
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&cond_);
}
static void* Routine(void* argv)//类内函数有this指针，将其设置为静态，argv接受this
{
pthread_detach(pthread_self());
threadpool<T> *tp = static_cast<threadpool<T>*>(argv);
while(true)
{
cout<<"pthread["<<pthread_self()<<"]running"<<endl;
sleep(1);
tp->lockQueue();
while(tp->isempty())
{
tp->waitTask();
}
T t=tp->pop();//拿到任务
tp->unlockQueue();
int one, two;
t.get(&one, &two);
cout<< "新线程 "<<pthread_self()<<" 完成计算任务: " << one << "+" << two << "=" << t.run() << "\n";
}
void start()//创建pthread_num个线程
{
assert(is_start);
for(int i=0;i<pthread_num;i++)
{
pthread_t tid;
pthread_create(&tid,nullptr,Routine,this);
}
is_start=false;
}
}
private:
//封装的接口
void lockQueue()
{
pthread_mutex_lock(&mutex_);
}
void unlockQueue()
{
pthread_mutex_unlock(&mutex_);
}
void waitTask()
{
pthread_cond_wait(&cond_,&mutex_);
}
void SignalTask()
{
pthread_cond_signal(&cond_);
}
bool isempty()
{
return _q.empty();
}
bool isFull()
{
return _q.size()==capacity;
}
private:
queue<T> _q;
pthread_mutex_t mutex_;//互斥锁
pthread_cond_t cond_;//信号量
int pthread_num;//创建线程数量
bool is_start;
};pop，push与执行任务的接口：
void push(const T x)
{
lockQueue();
_q.push(x);
unlockQueue();
SignalTask();
}
T pop()
{
T x=_q.front();
_q.pop();
return x;
}
将其改为单例模式：
ThreadPool(const ThreadPool<T> &) = delete;
void operator=(const ThreadPool<T>&) = delete;增加一静态成员变量，用来创建线程池，并初始为空。
static ThreadPool<T> *instance
template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;提供一个访问单例对象的函数
static ThreadPool<T> *getInstance()
{
if (nullptr == instance) //过滤重复的判断
{
if (nullptr == instance)
{
instance = new ThreadPool<T>();
}
}
return instance;
}任务函数：完成两数相加
class Task
{
public:
Task(int a = 10, int b = 9)
: a_(a), b_(b)
{
}
int run()
{
return a_ + b_;
}
void getTask(int *a, int *b)
{
*a = a_;
*b = b_;
}
private:
int a_;
int b_;
};
完整代码：
#include "Task.hpp"
#define pthnums 3
#include<unistd.h>
#include<queue>
#include<assert.h>
#include<pthread.h>
#include<iostream>
using namespace std;
template <class T>
class threadpool
{
public:
threadpool(const threadpool<T> &) = delete;
void operator=(const threadpool<T> &) = delete;
static threadpool<T> *getInstance()
{
if (nullptr == instance) //过滤重复的判断
{
if (nullptr == instance)
{
instance = new threadpool<T>;
}
}
return instance;
}
threadpool(int nums = pthnums)
{
pthread_num = nums;
pthread_mutex_init(&mutex_, nullptr);
pthread_cond_init(&cond_, nullptr);
is_start = true;
}
~threadpool()
{
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&cond_);
}
static void *Routine(void *argv) //类内函数有this指针，将其设置为静态，argv接受this
{
pthread_detach(pthread_self());
threadpool<T> *tp = static_cast<threadpool<T> *>(argv);
while (true)
{
cout << "pthread[" << pthread_self() << "]running" << endl;
tp->lockQueue();
while (tp->isempty())
{
tp->waitTask();
}
T t = tp->pop(); //拿到任务
tp->unlockQueue();
int one, two;
t.get(&one, &two);
cout << "新线程 " << pthread_self() << " 完成计算任务: " << one << "+" << two << "=" << t.run() << "\n";
}
}
void start() //创建pthread_num个线程
{
assert(is_start);
for (int i = 0; i < pthread_num; i++)
{
pthread_t tid;
pthread_create(&tid, nullptr, Routine, this);
}
is_start = false;
}
void push(const T x)
{
lockQueue();
_q.push(x);
unlockQueue();
SignalTask();
}
T pop()
{
T x = _q.front();
_q.pop();
return x;
}
private:
//封装的接口
void lockQueue()
{
pthread_mutex_lock(&mutex_);
}
void unlockQueue()
{
pthread_mutex_unlock(&mutex_);
}
void waitTask()
{
pthread_cond_wait(&cond_, &mutex_);
}
void SignalTask()
{
pthread_cond_signal(&cond_);
}
bool isempty()
{
return _q.empty();
}
private:
queue<T> _q;
pthread_mutex_t mutex_; //互斥锁
pthread_cond_t cond_; //信号量
int pthread_num; //创建线程数量
bool is_start;
static threadpool<T> *instance;
};
template <class T>
threadpool<T> *threadpool<T>::instance = nullptr;
进行测试：
#include "threadpool.hpp"
#include <time.h>
int main()
{
srand(time(nullptr));
threadpool<Task> *p = threadpool<Task>::getInstance();
int a;
int b;
p->start();
while (true)
{
a = rand() % 100;
b = rand() % 50;
Task t(a, b);
p->push(t);
//sleep(1);
}
return 0;
}
结果：
