线程池：线程池的实现 | 日志

原理

在一个可执行程序内部存在多个线程和一个任务队列。如果任务队列里长时间没有任务，这些线程就会休眠，如果此时来了一个任务，那么线程就会被唤醒。像这种，提前创建好线程，需要的时候直接使用，我们称之为线程池。这种本质上就是一个生产消费模型。

线程池实现

//ThreadPool.hpp
#pragma once#include<iostream>
#include<unistd.h>
#include<string>
#include<vector>
#include<queue>
#include<functional>
#include"Thread.hpp"using namespace threadModel;static const int gdefaultnum=5;void test()
{while(true){std::cout<<"hello world"<<std::endl;sleep(1);}
}template<typename T>
class ThreadPool
{
private:void LockQueue(){pthread_mutex_lock(&_mutex);}void UnlockQueue(){pthread_mutex_unlock(&_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}void WakeupAll(){pthread_cond_broadcast(&_cond);}void Sleep(){pthread_cond_wait(&_cond,&_mutex);}bool IsEmpty(){return _task_queue.empty();}void HandlerTask(const std::string& name)  // this{while (true){LockQueue();//如果队列为空(有任务)while(IsEmpty()&&_isrunning) //线程没有任务，但是在工作，继续休眠{_sleep_thread_num++;Sleep();_sleep_thread_num--;}if(IsEmpty()&&!_isrunning) // 任务是空的，并且线程退出工作{std::cout<<name<<" quit..."<<std::endl;UnlockQueue();break;}// 队列不为空，有任务 或者 队列被唤醒// 取任务T t=_task_queue.front();_task_queue.pop();UnlockQueue();// 此处任务已经不在任务队列中，任务已经被拿走，处理任务和临界资源是两码事t(); // 处理任务，不能不用也不能在临界区中处理std::cout<<name<<": "<<t.result()<<std::endl;}}public:ThreadPool(int thread_num=gdefaultnum):_thread_num(thread_num),_isrunning(false) //刚开始线程没有使用,_sleep_thread_num(0){pthread_mutex_init(&_mutex,nullptr);pthread_cond_init(&_cond,nullptr);}void Init(){func_t func=std::bind(&ThreadPool::HandlerTask,this,std::placeholders::_1);for(int i=0;i<_thread_num;i++){std::string threadname="thread-"+std::to_string(i+1);_threads.emplace_back(threadname,func);}}void Start(){_isrunning=true;for(auto& thread:_threads){thread.Start();}}void Stop(){LockQueue();_isrunning=false;WakeupAll();UnlockQueue();}void Equeue(const T &in){LockQueue();if(_isrunning){_task_queue.push(in);// 如果当前有线程在等待，需要唤醒if(_sleep_thread_num>0){Wakeup();}}UnlockQueue();}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}
private:int _thread_num;std::vector<Thread> _threads;  // 管理多个线程std::queue<T> _task_queue; // 任务队列bool _isrunning; //当前线程是否在工作int _sleep_thread_num;   //计数器：休眠的线程个数pthread_mutex_t _mutex;pthread_cond_t _cond;
};

日志

日志是软件运行的记录信息，可以向显示器打印，也可以向文件中打印，日志必须有特定的格式：

• [日志等级] [pid] [filename] [filenumber] [time] 日志内容（支持可变参数）

日志等级：DEBUG、INFO、WARNING、ERROR、FATAL（致命的）

// 日志等级
enum
{DEBUG=1,INFO,WARING,ERROR,FATAL
};

• 日志消息：日志等级、id、文件名、行号、当前时间等

// 日志消息
class logmessage
{public:std::string _level; // 日志等级pid_t _id; std::string _filename; // 文件名int _filenumber;  // 行号std::string _cur_time;std::string _message_info;};

获取当前时间函数接口

std::string GetCurTime()
{time_t now=time(nullptr);struct tm* cur_time=localtime(&now);char buffer[128];snprintf(buffer,sizeof(buffer),"%d-%02d-%02d %02d:%02d:%02d",cur_time->tm_year+1900,cur_time->tm_mon+1,cur_time->tm_mday,cur_time->tm_hour,cur_time->tm_min,cur_time->tm_sec);return std::string(buffer);
}

• time(nullptr) 返回当前的时间（从 1970 年 1 月 1 日到现在的秒数），并将其赋值给 now 变量。time_t 是表示时间点的类型。
• localtime(&now) 将 now 转换为当地时间，并返回一个指向 tm 结构的指针。tm 结构包含了年、月、日、时、分、秒等信息。
  

启用类型设置

void Enable(int type)
{_type=type;
}

Enable 函数用于设置日志输出类型，可以选择输出到屏幕或文件。

输出到屏幕

void FlushLogToSCreen(const logmessage& lg)
{printf("[%s][%d][%s][%d][%s] %s",lg._level.c_str(),lg._id,lg._filename.c_str(),lg._filenumber,lg._cur_time.c_str(),lg._message_info.c_str());
}

FlushLogToSCreen 函数将日志信息格式化并输出到控制台。使用 printf 格式化字符串。

输出到文件

void FlushLogToFile(const logmessage& lg)
{std::ofstream out(_logfile,std::ios::app); // 追加打印if(!out.is_open())return;char logtxt[2048];snprintf(logtxt,sizeof(logtxt),"[%s][%d][%s][%d][%s] %s",lg._level.c_str(),lg._id,lg._filename.c_str(),lg._filenumber,lg._cur_time.c_str(),lg._message_info.c_str());out.write(logtxt,strlen(logtxt));out.close();
}

FlushLogToFile 函数将日志信息写入指定的文件。以追加模式打开文件，并在打开失败时返回。
使用snprintf 格式化日志信息，然后将其写入文件。

选择输出方式

void FlushLog(const logmessage& lg)
{switch(_type){case SCREEN_TYPE:FlushLogToSCreen(lg);break;case FILE_TYPE:FlushLogToFile(lg);break;}
}

创建日志消息

void logMessage(std::string filename,int filenumber,int level,const char* format,...)
{logmessage lg;lg._level=LevelToString(level);lg._id=getpid();lg._filename=filename;lg._filenumber=filenumber;lg._cur_time=GetCurTime();va_list ap;va_start(ap,format);char log_info[1024];vsnprintf(log_info,sizeof(log_info),format,ap);va_end(ap);lg._message_info=log_info;FlushLog(lg);
}

• logMessage 函数用于创建一条新的日志消息。它接受文件名、文件编号、日志级别和格式化字符串作为参数。
• 使用可变参数处理（va_list）来处理格式化字符串。
• 将生成的日志信息存储在 lg 对象中，并调用 FlushLog 函数进行输出。
• va_list ap;声明了一个 va_list 类型的变量 ap，它用于存储可变参数列表。在 C 语言中，va_list 是一个用于遍历不定数量参数的类型。
• va_start(ap, format);：va_start 宏初始化 ap 以指向函数参数列表中的第一个可变参数。这里的 format 是最后一个固定参数，va_start 会从它的下一个参数开始读取可变参数。因此，ap 现在可以用来访问 format 之后的所有参数。
• va_end(ap)用于清理va_list 变量 ap。在读取完可变参数后，调用 va_end 是良好的实践，它可以释放与 ap 相关的资源（如果有的话）。

完整代码

#pragma once#include<iostream>
#include<string>
#include<cstring>
#include<sys/types.h>
#include<unistd.h>
#include<stdarg.h>
#include<ctime>
#include<fstream>
#include<pthread.h>
#include<cstdio>
#include"LockGuard.hpp"using std::cin;
using std::cout;
using std::endl;namespace log_ns
{// 日志等级enum{DEBUG=1,INFO,WARING,ERROR,FATAL};// 日志消息class logmessage{public:std::string _level; // 日志等级pid_t _id; std::string _filename; // 文件名int _filenumber;  // 行号std::string _cur_time;std::string _message_info;};pthread_mutex_t glock=PTHREAD_MUTEX_INITIALIZER; // 定义一个全局的锁std::string LevelToString(int level){LockGuard lockguard(&glock);switch(level){case DEBUG:return "DEBUG";case INFO:return "INFO";case WARING:return "WARING";case ERROR:return "ERROR";case FATAL:return "FATAL";default:return "UNKNOWN";}}std::string GetCurTime(){time_t now=time(nullptr);struct tm* cur_time=localtime(&now);char buffer[128];snprintf(buffer,sizeof(buffer),"%d-%02d-%02d %02d:%02d:%02d",cur_time->tm_year+1900,cur_time->tm_mon+1,cur_time->tm_mday,cur_time->tm_hour,cur_time->tm_min,cur_time->tm_sec);return std::string(buffer);}#define SCREEN_TYPE 1#define FILE_TYPE 2const std::string glogfile="./log.txt";// 日志class Log{public:Log(const std::string& logfeile=glogfile):_logfile(logfeile),_type(SCREEN_TYPE){}void Enable(int type){_type=type;}void FlushLogToSCreen(const logmessage& lg){printf("[%s][%d][%s][%d][%s] %s",lg._level.c_str(),lg._id,lg._filename.c_str(),lg._filenumber,lg._cur_time.c_str(),lg._message_info.c_str());}void FlushLogToFile(const logmessage& lg){std::ofstream out(_logfile,std::ios::app); // 追加打印if(!out.is_open())return;char logtxt[2048];snprintf(logtxt,sizeof(logtxt),"[%s][%d][%s][%d][%s] %s",lg._level.c_str(),lg._id,lg._filename.c_str(),lg._filenumber,lg._cur_time.c_str(),lg._message_info.c_str());out.write(logtxt,strlen(logtxt));out.close();}void FlushLog(const logmessage& lg){switch(_type){case SCREEN_TYPE:FlushLogToSCreen(lg);break;case FILE_TYPE:FlushLogToFile(lg);break;}}void logMessage(std::string filename,int filenumber,int level,const char* format,...){logmessage lg;lg._level=LevelToString(level);lg._id=getpid();lg._filename=filename;lg._filenumber=filenumber;lg._cur_time=GetCurTime();va_list ap;va_start(ap,format);char log_info[1024];vsnprintf(log_info,sizeof(log_info),format,ap);va_end(ap);lg._message_info=log_info;//cout<<lg._message_info<<endl;// 打印日志FlushLog(lg);}~Log(){}private:int _type;std::string _logfile;};Log lg;#define LOG(Level,Format,...) do{lg.logMessage(__FILE__,__LINE__,Level,Format,##__VA_ARGS__);}while(0)#define EnableScreen() do{lg.Enable(SCREEN_TYPE);}while(0)#define EnableFile() do{lg.Enable(FILE_TYPE);}while(0)};

携带日志的线程池设计

//ThreadPool.hpp
#pragma once#include<iostream>
#include<unistd.h>
#include<string>
#include<vector>
#include<queue>
#include<functional>
#include"Thread.hpp"
#include"Log.hpp"using namespace threadModel;
using namespace log_ns;static const int gdefaultnum=5;void test()
{while(true){std::cout<<"hello world"<<std::endl;sleep(1);}
}template<typename T>
class ThreadPool
{
private:void LockQueue(){pthread_mutex_lock(&_mutex);}void UnlockQueue(){pthread_mutex_unlock(&_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}void WakeupAll(){pthread_cond_broadcast(&_cond);}void Sleep(){pthread_cond_wait(&_cond,&_mutex);}bool IsEmpty(){return _task_queue.empty();}void HandlerTask(const std::string& name)  // this{while (true){LockQueue();//如果队列为空(有任务)while(IsEmpty()&&_isrunning) //线程没有任务，但是在工作，继续休眠{_sleep_thread_num++;LOG(INFO,"%s thread sleep begin!\n",name.c_str());Sleep();LOG(INFO,"%s thread wakeup!\n",name.c_str());_sleep_thread_num--;}if(IsEmpty()&&!_isrunning) // 任务是空的，并且线程退出工作{UnlockQueue();LOG(INFO,"%s quit\n",name.c_str());break;}// 队列不为空，有任务 或者 队列被唤醒// 取任务T t=_task_queue.front();_task_queue.pop();UnlockQueue();// 此处任务已经不在任务队列中，任务已经被拿走，处理任务和临界资源是两码事t(); // 处理任务，不能不用也不能在临界区中处理LOG(DEBUG,"hander task done, task is: \n%s",t.result().c_str());}}public:ThreadPool(int thread_num=gdefaultnum):_thread_num(thread_num),_isrunning(false) //刚开始线程没有使用,_sleep_thread_num(0){pthread_mutex_init(&_mutex,nullptr);pthread_cond_init(&_cond,nullptr);}void Init(){func_t func=std::bind(&ThreadPool::HandlerTask,this,std::placeholders::_1);for(int i=0;i<_thread_num;i++){std::string threadname="thread-"+std::to_string(i+1);_threads.emplace_back(threadname,func);LOG(DEBUG,"construct thread %s done, init success.\n",threadname.c_str());}}void Start(){_isrunning=true;for(auto& thread:_threads){LOG(DEBUG,"Start thread %s done.\n",thread.Name().c_str());thread.Start();}}void Stop(){LockQueue();_isrunning=false;WakeupAll();UnlockQueue();LOG(INFO,"Thread Pool Stop Success!\n");}void Equeue(const T &in){LockQueue();if(_isrunning){_task_queue.push(in);// 如果当前有线程在等待，需要唤醒if(_sleep_thread_num>0){Wakeup();}}UnlockQueue();}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}
private:int _thread_num;std::vector<Thread> _threads;  // 管理多个线程std::queue<T> _task_queue; // 任务队列bool _isrunning; //当前线程是否在工作int _sleep_thread_num;   //计数器：休眠的线程个数pthread_mutex_t _mutex;pthread_cond_t _cond;
};

//Main.cc
#include"ThreadPool.hpp"
#include"Task.hpp"
#include"Log.hpp"
#include<memory>using std::cin;
using std::cout;
using std::endl;
using namespace log_ns;int main()
{EnableScreen();//std::unique_ptr<ThreadPool> tp=std::make_unique<>();  //构建一个ThreadPool对象ThreadPool<Task> *tp=new ThreadPool<Task>();tp->Init();tp->Start();int cnt=10;while (cnt){// 不断地向线程池推送任务sleep(1);Task t(1,1);tp->Equeue(t);LOG(INFO,"equeue a task, %s\n",t.debug().c_str());sleep(1);}tp->Stop();LOG(INFO,"thread pool stop!\n");sleep(10);return 0;
}

// Thread.hpp
#pragma once
#include <pthread.h>
#include <iostream>
#include <string>
#include<functional>namespace threadModel
{// 线程执行的方法//typedef void (*func_t)(ThreadData* td);using func_t=std::function<void(const std::string&)>;class Thread{public:void Excute(){_isrunning = true;_func(_name);_isrunning = false;}public:Thread(const std::string &name, func_t func) : _name(name), _func(func){}// void *ThreadRoutine(void* args)实际上参数里面还有一个Thread* thisstatic void *ThreadRoutine(void *args) // 加上static后，参数里面就没有Thread* this{Thread *self = static_cast<Thread *>(args); // 获得当前对象self->Excute();return nullptr;}bool Start(){int n = ::pthread_create(&_tid, nullptr, ThreadRoutine, this);if (n != 0)return false;return true;}std::string Status(){if (_isrunning)return "running";elsereturn "sleep";}void Stop(){if (_isrunning){pthread_cancel(_tid);_isrunning = false;}}void Join(){pthread_join(_tid, nullptr);}std::string Name(){return _name;}~Thread(){Stop();}private:std::string _name;pthread_t _tid;bool _isrunning;func_t _func; // 线程执行的回调函数};
}

//Task.hpp
#pragma once
#include<iostream>
#include<string>
#include<functional>class Task
{public:Task(){}Task(int x,int y):_x(x),_y(y){}void Excute(){_result=_x+_y;}void operator()(){Excute();}std::string debug(){std::string msg=std::to_string(_x)+"+"+std::to_string(_y)+"=?";return msg;}std::string result(){std::string msg=std::to_string(_x)+"+"+std::to_string(_y)+"="+std::to_string(_result);return msg;}private:int _x;int _y;int _result;
};