Linux线程池
- Linux线程池
- 线程池的概念
- 线程池的优点
- 线程池的应用场景
- 线程池的实现
Linux线程池
线程池的概念
线程池是一种线程使用模式。
线程过多会带来调度开销，进而影响缓存局部和整体性能，而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。
线程池的优点
- 线程池避免了在处理短时间任务时创建与销毁线程的代价。
- 线程池不仅能够保证内核充分利用，还能防止过分调度。
注意： 线程池中可用线程的数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
线程池的应用场景
线程池常见的应用场景如下：
- 需要大量的线程来完成任务，且完成任务的时间比较短。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。
相关解释：
- 像Web服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。
- 对于长时间的任务，比如Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
- 突发性大量客户请求，在没有线程池的情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，但短时间内产生大量线程可能使内存到达极限，出现错误。
线程池的实现
下面我们实现一个简单的线程池，线程池中提供了一个任务队列，以及若干个线程（多线程）。
- 线程池中的多个线程负责从任务队列当中拿任务，并将拿到的任务进行处理。
- 线程池对外提供一个Push接口，用于让外部线程能够将任务Push到任务队列当中。
- 将线程池设计为单例对象(懒汉模式实现)
线程池的代码如下：
#include "lockGuard.hpp"
#include "thread.hpp"
#include <vector>
#include <string>
#include <queue>
#include <iostream>
#include <unistd.h>
#include "log.hpp"
const int g_thread_num = 3;
//单例线程池(懒汉模式)
template <class T>
class ThreadPool
{
private:
std::vector<Thread *> _threads;//储存线程
int _num; //线程池中线程的数量
std::queue<T> _taskQueue; //任务队列
pthread_mutex_t _mutex; // 互斥锁
pthread_cond_t _cond; // 条件变量
static ThreadPool<T>* _threadPoolPtr;//单例对象指针
static pthread_mutex_t _mtx;//单例对象的互斥锁
private:
//构造函数私有化
ThreadPool(int thread_num = g_thread_num) : _num(thread_num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for (int i = 0; i < _num; ++i)
{
//在构造线程对象时,将单例线程池对象的指针传给线程对象
_threads.push_back(new Thread(i, routine, this));
}
}
//防拷贝
ThreadPool(const ThreadPool<T>& threadpool) = delete;
ThreadPool<T> operator=(const ThreadPool<T>& threadpool) = delete;
public:
pthread_mutex_t *getMutex()
{
return &_mutex;
}
void waitCond()
{
pthread_cond_wait(&_cond, &_mutex);
}
public:
//全局访问点
static ThreadPool<T>* getThreadPool(int num = g_thread_num)
{
if(_threadPoolPtr == nullptr)
{
pthread_mutex_lock(&_mtx);
if(_threadPoolPtr == nullptr)
{
_threadPoolPtr = new ThreadPool<T>(num);
}
pthread_mutex_unlock(&_mtx);
}
return _threadPoolPtr;
}
//启动线程
void run()
{
for (auto &it : _threads)
{
it->start();
//std::cout << it->name() << " 启动成功" << std::endl;
logMessage(NORMAL,"%s %s",it->name().c_str(),"启动成功 ");
}
}
T getTask()
{
T task = _taskQueue.front();
_taskQueue.pop();
return task;
}
// 消费过程
//将routine函数设为静态,是因为routine是作为线程例程函数的,所以如果不将routine设为静态,routine函数的参数就会多出一个this指针,无法作为线程例程
static void *routine(void *args)
{
ThreadData *td = (ThreadData *)args;
ThreadPool<T> *tp = (ThreadPool<T> *)td->args_;
while (true)
{
T task;
{
lockGuard lock(tp->getMutex());//构造加锁,出作用域解锁
while (tp->_taskQueue.empty())
tp->waitCond();
// 读取任务
task = tp->getTask(); // 任务队列是共享的-> 将任务从共享，拿到自己的私有空间
}
// 处理任务
task(td->name_);
// sleep(1);
}
}
// 生产过程
void pushTask(const T &task)
{
lockGuard lock(&_mutex);
_taskQueue.push(task);
pthread_cond_signal(&_cond);
}
~ThreadPool()
{
for (auto &it : _threads)
{
it->join();
delete it;
}
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
};
template <class T>
ThreadPool<T>* ThreadPool<T>::_threadPoolPtr = nullptr;
//静态初始化
template <class T>
pthread_mutex_t ThreadPool<T>::_mtx = PTHREAD_MUTEX_INITIALIZER;
为什么线程池中需要有互斥锁和条件变量？
线程池中的任务队列是会被多个执行流同时访问的临界资源，因此我们需要引入互斥锁对任务队列进行保护。
线程池当中的线程要从任务队列里拿任务，前提条件是任务队列中必须要有任务，因此线程池当中的线程在拿任务之前，需要先判断任务队列当中是否有任务，若此时任务队列为空，那么该线程应该进行等待，直到任务队列中有任务时再将其唤醒，因此我们需要引入条件变量。
当外部线程向任务队列中Push一个任务后，此时可能有线程正处于等待状态，因此在新增任务后需要唤醒在条件变量下等待的线程。
注意：
- 当某线程被唤醒时，其可能是被异常或是伪唤醒，或者是一些广播类的唤醒线程操作而导致所有线程被唤醒，使得在被唤醒的若干线程中，只有个别线程能拿到任务。此时应该让被唤醒的线程再次判断是否满足被唤醒条件，所以在判断任务队列是否为空时，应该使用while进行判断，而不是if。
- pthread_cond_broadcast函数的作用是唤醒条件变量下的所有线程，而外部可能只Push了一个任务，我们却把全部在等待的线程都唤醒了，此时这些线程就都会去任务队列获取任务，但最终只有一个线程能得到任务。一瞬间唤醒大量的线程可能会导致系统震荡，这叫做惊群效应。因此在唤醒线程时最好使用pthread_cond_signal函数唤醒正在等待的一个线程即可。
- 当线程从任务队列中拿到任务后，该任务就已经属于当前线程了，与其他线程已经没有关系了，因此应该在解锁之后再进行处理任务，而不是在解锁之前进行。因为处理任务的过程可能会耗费一定的时间，所以我们不要将其放到临界区当中。
- 如果将处理任务的过程放到临界区当中，那么当某一线程从任务队列拿到任务后，其他线程还需要等待该线程将任务处理完后，才有机会进入临界区。此时虽然是线程池，但最终我们可能并没有让多线程并行的执行起来。
为什么线程池中的线程执行例程需要设置为静态方法？
使用pthread_create函数创建线程时，需要为创建的线程传入一个Routine（执行例程），该Routine只有一个参数类型为void的参数，以及返回类型为void的返回值。
而此时Routine作为类的成员函数，该函数的第一个参数是隐藏的this指针，因此这里的Routine函数，虽然看起来只有一个参数，而实际上它有两个参数，此时直接将该Routine函数作为创建线程时的执行例程是不行的，无法通过编译。
静态成员函数属于类，而不属于某个对象，也就是说静态成员函数是没有隐藏的this指针的，因此我们需要将Routine设置为静态方法，此时Routine函数才真正只有一个参数类型为void*的参数。
但是在静态成员函数内部无法调用非静态成员函数，而我们需要在Routine函数当中调用该类的某些非静态成员函数，比如Pop。因此我们需要在创建线程时，向Routine函数传入的当前对象的this指针，此时我们就能够通过该this指针在Routine函数内部调用非静态成员函数了。
线程封装
#include <iostream>
#include <cstdio>
#include <pthread.h>
typedef void*(*fun_t)(void*);//函数指针
class ThreadData
{
public:
void *args_;
std::string name_;
};
//对线程进行封装
class Thread
{
private:
std::string _name; //线程名字
fun_t _func; //线程例程
ThreadData _tdata; //线程例程的参数
pthread_t _tid; //线程id
public:
Thread(int num,fun_t callback,void* args):_func(callback)
{
char nameBuffer[64];
snprintf(nameBuffer,sizeof(nameBuffer),"Thread-%d",num);
_name = nameBuffer;
_tdata.args_ = args;
_tdata.name_ = _name;
}
//创建线程
void start()
{
pthread_create(&_tid,nullptr,_func,(void*)&_tdata);
}
//等待线程
void join()
{
pthread_join(_tid,nullptr);
}
std::string name()
{
return _name;
}
~Thread()
{}
};
任务类型的设计
我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的，但无论该任务是什么类型的，在该任务类当中都必须包含一个Run方法，当我们处理该类型的任务时只需调用该Run方法即可。
typedef std::function<int(int,int,char)> Fun_t;
class Task
{
private:
int _x;
int _y;
char _op;
Fun_t _func;
public:
Task() = default;
Task(int x,int y,char op,Fun_t func)
:_x(x),_y(y),_op(op),_func(func)
{ }
void operator()(const std::string& name)
{
//std::cout << "线程 " << name << " 处理完成, 结果是: " << _x << _op << _y << "=" << _func(_x, _y,_op) << std::endl;
//__FILE__是一个预定义符号,表示进行编译的文件的名字,
//__LINE__表示文件当前的行号
logMessage(NORMAL,"%s %d+%d=%d | %s | %d",(char*)name.c_str(),_x,_y,_func(_x,_y,_op),__FILE__,__LINE__);
}
};
互斥锁的封装以及RAII风格的加锁
//互斥锁的封装
class Mutex
{
private:
pthread_mutex_t* pmtx_;
public:
Mutex(pthread_mutex_t* pmtx):pmtx_(pmtx)
{ }
void lock()
{
pthread_mutex_lock(pmtx_);
}
void unlock()
{
pthread_mutex_unlock(pmtx_);
}
~Mutex()
{ }
};
class lockGuard
{
private:
/* data */
Mutex _mtx;
public:
lockGuard(pthread_mutex_t* pmtx);
~lockGuard();
};
lockGuard::lockGuard(pthread_mutex_t* pmtx):_mtx(pmtx)
{
_mtx.lock();
}
lockGuard::~lockGuard()
{
_mtx.unlock();
}
日志功能
//日志等级
#define DEBUG 0 //调试
#define NORMAL 1 //正常
#define WARNING 2 //警告
#define ERROR 3 //错误但不影响后续的执行
#define FATAL 4 //致命(严重错误)无法继续运行
#define LOGFILE "./log.txt"
const char* gLevelMap[] = {
"DEBUG",
"NORMAL",
"WARNING",
"ERROR",
"FATAL",
};
//完整的日志功能,至少: 日志等级 时间 日志内容 支持用户自定义
void logMessage(int level, char* format,...)
{
//提取可变参数
// va_list ap;
// va_start(ap,format);
//while()
// va_end(ap);
#ifndef DEBUG_SHOW
if(level == DEBUG) return;
#endif
char stdBuffer[1024];//标准部分
time_t timep;
time(&timep);
snprintf(stdBuffer,sizeof(stdBuffer),"%s %s",gLevelMap[level],ctime(&timep));
stdBuffer[strlen(stdBuffer)-1] = '\0';
char logBuffer[1024];//自定义部分
va_list ap;
va_start(ap,format);
// vprintf(format,ap);
vsnprintf(logBuffer,sizeof(logBuffer),format,ap);
va_end(ap);
FILE* fp = fopen(LOGFILE,"a");
if(fp == nullptr)
{
perror("fopen");
}
//printf("%s %s\n",stdBuffer,logBuffer);
fprintf(fp,"%s %s\n",stdBuffer,logBuffer);
fclose(fp);
}
主逻辑
#include "threadPool.hpp"
#include <iostream>
#include "task.hpp"
static char op[4] = {'+','-','*','/'};
int calculate(int x,int y,char op)
{
switch(op)
{
case '+' : return x + y;
break;
case '-' : return x - y;
break;
case '*' : return x * y;
break;
case '/' : return x / y;
break;
}
}
int main()
{
srand(time(nullptr));
//ThreadPool<Task>* tp = new ThreadPool<Task>();
ThreadPool<Task>* tp = ThreadPool<Task>::getThreadPool(5);
tp->run();
while(true)
{
//生产的过程，制作任务的时候，要花时间
int x = rand() % 100;
int y = rand() % 100 + 1;
int index = rand() % 4;
Task task(x,y,op[index],calculate);
//std::cout << "制作任务完成: " << x << op[index] << y << "=?" << std::endl;
logMessage(DEBUG,"制作任务完成: %d%c%d=?",x,op[index],y);
tp->pushTask(task);
sleep(1);
}
return 0;
}
