引言
因为多线程对于CPU的高效利用 好几种高性能的服务器框架都使用了多线程 但线程的创建和回收是非常浪费系统资源的 常常会有不必要的时间损失 但我们的服务器的硬件确相对来说非常充裕 于是我们可以初始化一组资源 在服务器运行阶段可以直接获取而不需要重新分配 相对的在一个逻辑单元执行完以后也不需要释放资源 从而大大提高了效率 避免了对内核的频繁访问 这里我们说的资源就是线程 初始化的这一组资源可以看做是一个线程池
线程池基本原理
其实线程池的内部实现就相当于是一个生产者消费者模型 线程池就是消费者 同时竞争锁 先拿到锁则获取资源 生产者则相当于我们的append操作 我们每添加一次资源 则相当于生产者生产 内部则触发一个条件变量的 signal 唤醒工作线程 下面是一个简单的实现 其中把条件变量和线程池分别封装成不同的类 方便使用
ThreadPool.h
#ifndef THREADPOOL_H_
#define THREADPOOL_H_
#include"locker.h"
#include<list>
#include<cstdio>
#include<exception>
#include<pthread.h>
#include<iostream>
template<typename T>
void solve(const T &t){
std::cout << "线程接收到消息: " << t << std::endl;
return;
}
template<typename T>
class threadpool{
enum{Max_Thread = 10,Max_Requests = 10000};
public:
explicit threadpool(int thread_number = Max_Thread,int max_requests = Max_Requests);
~threadpool();
bool append(T* request);//向任务队列中添加任务
private:
static void* worker(void *arg);
void run();
private:
int m_thread_number;//最大线程数
int m_max_requests;//最大请求数
pthread_t* m_threads; //描述线程的数组
//unique_ptr<pthread_t[]>m_threads; //如果确定数组大小 就可以用unique_ptr管理内存
std::list<T*> m_workqueue;//任务队列
cond m_QueueState; //条件变量 类内初始化
bool m_stop;
//直接类内默认初始化失败
};
template<typename T>
threadpool<T>::threadpool(int thread_number,int max_requests):
m_thread_number(thread_number),m_max_requests(max_requests),m_stop(false),m_threads(NULL),
m_QueueState(){//默认初始化
if(thread_number<=0 || max_requests<=0){
throw std::out_of_range("error in initialize");
}
m_threads = new pthread_t[m_thread_number];
if(!m_threads){
throw std::bad_alloc();
}
//typedef void* (*Temp)(void*);
using Temp = void* (*)(void*);
Temp enp = (Temp)&threadpool::run;
for(int i=0;i<m_thread_number;++i){
if(pthread_create(&(m_threads[i]),nullptr,enp,this) != 0){
//if(pthread_create(&(m_threads[i]),nullptr,worker,this) != 0){ //两种都可以
delete []m_threads;
throw std::exception();
}
if(int ret = pthread_detach(m_threads[i])){
delete []m_threads;
throw std::exception();
}
}
}
template<typename T>
threadpool<T>::~threadpool(){
delete []m_threads;
m_stop = true;
}
template<typename T>
bool threadpool<T>::append(T* requests){
m_QueueState.lock();
if(m_workqueue.size() >= m_max_requests){
m_QueueState.unlock();
return false;
}
m_workqueue.emplace_back(requests);
m_QueueState.signal();
m_QueueState.unlock();
return true;
}
template<typename T> //这个函数本来为pthread_create的参数准备的 但有更好的解决方案 及强制类型转换
void* threadpool<T>::worker(void *arg){
threadpool *pool = (threadpool*)arg;
pool->run();
return pool;
}
template<typename T>
void threadpool<T>::run(){
while(!m_stop){
m_QueueState.lock();
while(m_workqueue.empty()){
m_QueueState.wait();
}
T* request = m_workqueue.front();
m_workqueue.pop_front();
m_QueueState.unlock();
//这里就是线程得到这个参数后如何执行
std::cout << pthread_self() << std::endl;
solve(*request);
}
}
#endif
locker.h
#ifndef LOCKER_H_
#define LOCKER_H_
#include<exception>
#include<pthread.h>
#include<semaphore.h>//信号量
class locker{
public:
locker(){
if(pthread_mutex_init(&m_mutex,nullptr)!=0){
throw std::exception();
}
}
~locker(){
pthread_mutex_destroy(&m_mutex);
}
bool lock(){
return pthread_mutex_lock(&m_mutex) == 0;
}
bool unlock(){
return pthread_mutex_unlock(&m_mutex) == 0;
}
private:
pthread_mutex_t m_mutex;
};
class cond{
public:
cond(){
if(pthread_mutex_init(&m_mutex,nullptr) != 0){
throw std::exception();
}
if(pthread_cond_init(&m_cond,nullptr) != 0){
pthread_mutex_destroy(&m_mutex);
throw std::exception();
}
}
~cond(){
pthread_mutex_destroy(&m_mutex);
pthread_cond_destroy(&m_cond);
}
bool signal(){
return pthread_cond_signal(&m_cond) == 0;
}
bool wait(){
int ret = pthread_cond_wait(&m_cond, &m_mutex);
return ret == 0;
}
bool lock(){
return pthread_mutex_lock(&m_mutex) == 0;
}
bool unlock(){
return pthread_mutex_unlock(&m_mutex) == 0;
}
private:
pthread_mutex_t m_mutex;
pthread_cond_t m_cond;
};
#endif
test_for_ThreadPool.cpp
#include"ThreadPool.h"
#include<iostream>
#include<unistd.h>
using namespace std;
int flag,tmp;
int main(){
threadpool<int> Pool(5,20000);
while(1){
//有可能超过最大请求数
tmp = ++flag;
//Pool.append(&tmp);
//加与不加while循环效率相差几十倍 可以测试下 奇怪
while(!Pool.append(&tmp)){
//cout << "重 11111111111111111111111111111111111111111复\n"; return 0;
};//加while的原因是把超过最大请求的重新发送
//sleep(2);
}
return 0;
}