网硕互联帮助中心Linux线程池
线程池
日志
设计模式：是软件开发领域，针对反复出现的经典业务场景，总结沉淀下来的成熟、可复用的通用解决方案，用来提高代码的可维护和可扩展性。
策略模式：将可变的算法 / 行为单独封装，和使用算法的主体解耦，可以在运行时动态切换不同算法行为。
日志可以输出到控制台、文件、网络。如果不使用策略模式，则日志类内部会有大量 if‑else 判断。每新增一种输出方式，比如输出到数据库，就必须修改日志类的源码，违反开闭原则（对扩展开放，对修改关闭）。
可以用策略模式改造日志类。
写一个父类（抽象策略）
做一个公共父类，把日志输出这个行为定义成纯虚函数。只要继承这个父类的子类，必须重写这个输出函数，保证所有输出方式都遵守同一套函数接口。
写多个子类（具体策略）
每一种日志输出方式，单独写一个子类去继承上面的父类。比如控制台输出、文件输出，分别做成不同子类。每个子类里面，只写自己专属的输出代码。
写日志主类（Context 上下文）
日志主类里面保存一份父类的指针 / 引用，不再写 if‑else 判断。对外提供修改这个指针的接口，运行的时候可以换成任意一个子类对象。当要打印日志的时候，直接调用指针的输出函数，依靠多态自动执行对应子类的代码。
以后新增一种日志输出方式，只需要再写一个新的子类继承父类并重写函数，完全不用修改日志主类的代码。
策略模式优点：遵循开闭原则，新增算法，不用修改原有代码；消除大量 if‑else、switch 分支判断；运行时动态切换算法策略（多态）；算法实现独立，便于单元测试。缺点：每一个不同策略，都要新增一个类，类数量膨胀；客户端必须理解各个策略差异，才知道选哪个策略；策略和上下文之间会多一层对象交互。
实现日志类，日志类的输出方式有往显示器输出和往文件输出，定义基类 LogStrategy ，公共接口为输出方式 SyncLog() ，派生类 ConsoleLogStrategy 和 FileLogStrategy 分别实现这两种输出方式。
//Mutex.hpp
#pragma once
#include <iostream>
#include <pthread.h>
#include <cstring>
namespace MutexModule
{
class Mutex
{
public:
Mutex()
{
int n = pthread_mutex_init(&_mutex,nullptr);
if(n!=0)
std::cout<<"Mutex() "<<strerror(n)<<std::endl;
}
~Mutex()
{
int n = pthread_mutex_destroy(&_mutex);
if(n!=0)
std::cout<<"~Mutex() "<<strerror(n)<<std::endl;
}
void Lock()
{
int n = pthread_mutex_lock(&_mutex);
if(n!=0)
std::cout<<"Lock() "<<strerror(n)<<std::endl;
}
void UnLock()
{
int n = pthread_mutex_unlock(&_mutex);
if(n!=0)
std::cout<<"UnLock() "<<strerror(n)<<std::endl;
}
pthread_mutex_t* GetAddr() {return &_mutex;}
private:
pthread_mutex_t _mutex;
};
class LockGuard
{
public:
LockGuard(Mutex* m)
:_m(m)
{
_m->Lock();
}
~LockGuard()
{
_m->UnLock();
}
private:
Mutex* _m;
};
}
//Log.hpp
#ifndef __LOG_HPP__
#define __LOG_HPP__
#include <iostream>
#include <string>
#include <unistd.h>
#include <memory>
#include <filesystem> //C++17
#include <sstream>
#include <fstream>
#include "Mutex.hpp"
namespace LogModule
{
const std::string gsep = "\\r\\n";
//基类
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string& message) = 0;//纯虚函数，强制派生类重写
};
//派生类: 往显示器打印日志
class ConsoleLogStrategy : public LogStrategy
{
public:
ConsoleLogStrategy()
{}
~ConsoleLogStrategy()
{}
void SyncLog(const std::string& message) override //检查是否重写
{
MutexModule::LockGuard lg(&_mutex);
std::cout<<message<<gsep;
}
private:
MutexModule::Mutex _mutex;
};
//派生类: 往文件中打印日志
const std::string defaultpath = "./log";
const std::string defaultfilename = "my.log";
class FileLogStrategy : public LogStrategy
{
public:
FileLogStrategy(const std::string& path = defaultpath,const std::string& filename = defaultfilename)
:_path(path)
,_filename(filename)
{
if(std::filesystem::exists(_path)) //路径对应的文件/文件夹是否存在
return;
try
{
std::filesystem::create_directories(_path);//创建路径
}
catch(const std::filesystem::filesystem_error& e)
{
std::cerr<<e.what()<<"\\n";
}
}
~FileLogStrategy()
{}
void SyncLog(const std::string& message) override
{
MutexModule::LockGuard lg(&_mutex);
std::string file = _path + (_path.back() == '/' ? "" : "/") + _filename;
std::ofstream out(file,std::ios::app); //每次追加式写入
if(!out.is_open())
{
std::cout<<"创建"<<file<<"失败"<<std::endl;
return;
}
out<<message<<gsep;
out.close();
}
private:
std::string _path;
std::string _filename;
MutexModule::Mutex _mutex;
};
}
#endif
对于派生类 FileLogStrategy ，往文件中打印日志需要确定路径，<filesystem> 的接口 std::filesystem::exists() 可以判断路径是否存在，存在返回true，不存在返回false。std::filesystem::create_directories() 可以创建多级目录，如果创建失败则会抛异常 std::filesystem::filesystem_error 。所以类 FileLogStrategy 的构造函数的执行逻辑是初始化成员 _path 和_filename，如果路径 _path 不存在则新建路径。
对于 FileLogStrategy 的日志输出函数 SyncLog() ，把路径 _path 和文件名 _filename拼接成一个完整文件路径 file ； <filesystem> 常与文件流 <fstream> 配合使用，创建 ofstream 对象 out ，以 ios::app 追加写入的方式打开文件，调用 out 成员函数 is_open() 判断是否成功打开文件；ofstream 也重载了运算符 << ，能像标准输出一样使用，把日志 message 输出到文件中；最后，像C的文件操作一样，需要调用 close() 关闭文件。
日志输出函数 SyncLog() 需要加锁，避免多线程环境下，多个线程同时打印造成输出错乱。
基类 LogStrategy 的析构建议加 virtual ，防止内存泄漏。同时因为是抽象类，抽象类无法实例化出对象，所以不需要构造函数。
派生类重写的虚函数自动继承 virtual 属性，可以不写 virtual ，用 override 修饰，自动在编译期检查是否真的重写了基类虚函数。
我们希望日志输出的格式：
[可读性很好的时间] [日志等级] [进程pid] [打印对应日志的文件名][行号] – 消息内容(支持可变参数)
日志等级的定义如下。
namespace LogModule
{
enum class LogLevel //强类型枚举
{
DEBUG, //调试信息
INFO, //普通信息
WARMING,//警告
ERROR, //错误
FATAL //致命错误
};
std::string LevelToStr(LogLevel level) //日志等级转字符串
{
switch(level)
{
case LogLevel::DEBUG:
return "DEBUG";
case LogLevel::ERROR:
return "ERROR";
case LogLevel::FATAL:
return "FATAL";
case LogLevel::INFO:
return "INFO";
case LogLevel::WARMING:
return "WARMING";
default:
return "UNKNOW";
}
}
}
日志等级的定义使用强类型枚举 enum class ，枚举名字不会泄露到外层作用域，不会隐式转 int。
最终是要以字符串的形式输出到显示器或文件，所以封装一个函数 LevelToStr()，把日志等级转成字符串。
封装 GetTimeStamp() 获取时间。
namespace LogModule
{
std::string GetTimeStamp() //获取时间
{
time_t curtime = time(nullptr);
struct tm curtm;
localtime_r(&curtime,&curtm);
char timebuffer[64];
snprintf(timebuffer,sizeof(timebuffer),"%4d-%02d-%02d %02d:%02d:%02d",
curtm.tm_year+1900,
curtm.tm_mon+1,
curtm.tm_mday,
curtm.tm_hour,
curtm.tm_min,
curtm.tm_sec
);
return timebuffer;
}
}
C语言的 time() 的返回类型是 time_t ，本质是长整型，time() 返回的是从1970-01-01 00:00:00到现在的秒数，用变量 curtime 接收 time() 返回的时间戳。定义结构体 struct tm 对象，存放 localtime_r() 分解出的具体的年月日时分秒，localtime_r(&curr, &curr_tm) 是可重入线程安全版本，由调用者自己提供 curtm 栈上内存，没有共享静态变量，多线程下比 localtime() 更好。调用 snprintf() 把时间格式化为字符串保存到 timebuffer[] 。
实现日志主类 Logger 。
namespace LogModule
{
class Logger
{
private:
std::unique_ptr<LogStrategy> _fflush_strategy;
public:
Logger()
{
EnableConsoleStrategy();
}
void EnableConsoleStrategy()
{
_fflush_strategy = std::make_unique<ConsoleLogStrategy>();
}
void EnableFileStrategy()
{
_fflush_strategy = std::make_unique<FileLogStrategy>();
}
~Logger()
{}
class LogMessage
{
};
LogMessage operator()()
{
}
};
}
Logger 的成员变量 _fflush_strategy 应该是基类指针，要实现多态效果，必须是基类的指针或引用，因为只有基类的指针或引用才能既指向基类对象又指向派生类对象。使用智能指针可以自动释放内存，不用手动 new/delete，同时在日志类场景下，不需要拷贝，所以使用 unique_ptr 即可。
_fflush_strategy 通过 std::make_unique<T>(args) 正式分配内存。make_unique 返回 unique_ptr<T>，直接给 _fflush_strategy 赋值派生类指针，派生类对象可以通过切片赋值基类指针或引用。make_unique 相对于直接 new 更加安全，堆上开空间异常时自动释放。
在主函数中，通过调用 EnableConsoleStrategy() 和 EnableFileStrategy() 改变 _fflush_strategy 的指向，根据需要选用日志输出方式。
//make_unique<类名>(构造参数)
_fflush_strategy = std::make_unique<ConsoleLogStrategy>();
//二者等价
_fflush_strategy = std::unique_ptr<Strategy>(new ConsoleLogStrategy());
内部类 LogMessage 和 Logger 的 () 重载。
namespace LogModule
{
class Logger
{
class LogMessage
{
public:
LogMessage(LogLevel& level,const std::string& srcname,int line,Logger& logger)
:_level(level)
,_srcname(srcname)
,_line_num(line)
,_logger(logger)
,_cur_time(GetTimeStamp())
,_pid(getpid())
{
std::stringstream ss;
ss<<"["<<_cur_time<<"] "
<<"["<<LevelToStr(_level)<<"] "
<<"["<<_pid<<"] "
<<"["<<_srcname<<"] "
<<"["<<_line_num<<"] "
<<"- ";
_loginfo = ss.str();
}
~LogMessage()
{
if(_logger._fflush_strategy) //不为空，则输出日志
_logger._fflush_strategy->SyncLog(_loginfo);
}
// LogMessage(…) << "hell world" << "XXXX" << 3.14 << 1234
template <class T>
LogMessage& operator<<(const T& info)
{
std::stringstream ss;
ss<<info;
_loginfo += ss.str();
return *this; //以便支持类似连续赋值
}
private:
std::string _cur_time;//打印日志的时间
LogLevel _level; //日志等级
pid_t _pid; //进程pid
std::string _srcname; //打印对应日志的⽂件名
int _line_num; //行号
std::string _loginfo; //完整的日志信息
Logger& _logger;
};
LogMessage operator()(LogLevel level,const std::string& srcname,int line)
{
return LogMessage(level,srcname,line,*this);
}
};
}
内部类 LogMessage 的任务是把日志的各项信息拼接到一起，保存在成员 _loginfo 中。LogMessage 的构造使用 <sstream> 的 stringstream 类对象 ss，将日志各项信息按照格式插入到 ss 中，调用 stringstream 的成员函数 str() 取得拼接后的字符串，实现字符串的拼接目的。
LogMessage 的成员 _logger 是外部类 Logger 的引用，当 LogMessage 析构时，访问 _logger 的 _fflush_strategy ，传入一条完整日志 _loginfo ，执行指定日志输出策略。
Logger 重载 () ，通过调用 operator() ，传入日志等级、文件名、行号，得到 LogMessage 临时对象。对this指针解引用，得到 Logger 对象，可绑定 Logger& 。
我们期望这样使用日志类：
Logger logger;
logger(level,srcname,line)<<"消息内容"<<123<<3.14;
那么 LogMessage 还需实现 << 重载。operator<< 应该是模板，方便各种类型的插入。使用 stringstream ，把日志的消息内容插入到 stringstream 对象，再用 str() 方法转成字符串，拼接到字符串成员 _loginfo 。最后返回 LogMessage 的引用，实现类似连续赋值的效果，即如上的连续插入。
最后，logger(level,srcname,line) 创建 LogMessage 临时对象，这个临时对象紧接着又调用自己的成员函数 operator<< ，插入 "消息内容" ，此时再度返回这个临时对象，继续调用 operator<< 。当这一行结束，临时对象生命周期也结束，调用 LogMessage 析构，把完整日志输出的显示器或指定文件。
namespace LogModule
{
const std::string gsep = "\\r\\n";
class LogStrategy
{};
class ConsoleLogStrategy : public LogStrategy
{};
const std::string defaultpath = "./log";
const std::string defaultfilename = "my.log";
class FileLogStrategy : public LogStrategy
{};
enum class LogLevel
{};
std::string LevelToStr(LogLevel level)
{}
std::string GetTimeStamp()
{}
class Logger
{
class LogMessage
{};
};
Logger logger;
//宏
#define LOG(level) logger(level,__FILE__,__LINE__)
#define ENABLE_CONSOLE_LOGSTRATEGY() logger.EnableConsoleStrategy()
#define ENABLE_FILE_LOGSTRATEGY() logger.EnableFileStrategy()
}
在命名空间 LogModule 中提前创建好 Logger 对象 logger ，Main.cc 中包含头文件 Log.hpp ，这样 logger 就在 Main.cc 中全局有效，方便用户操作。再进一步封装宏，使操作更加方便，宏 __FILE__ 和 __LINE__ 可以直接得到当前文件和行号。
Log.hpp 总的代码如下：
#ifndef __LOG_HPP__
#define __LOG_HPP__
#include <iostream>
#include <string>
#include <unistd.h>
#include <memory>
#include <filesystem>
#include <sstream>
#include <fstream>
#include "Mutex.hpp"
namespace LogModule
{
const std::string gsep = "\\r\\n";
//基类
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string& message) = 0;//纯虚函数，强制派生类重写
};
//派生类: 往显示器打印日志
class ConsoleLogStrategy : public LogStrategy
{
public:
ConsoleLogStrategy()
{}
~ConsoleLogStrategy()
{}
void SyncLog(const std::string& message) override //检查是否重写
{
MutexModule::LockGuard lg(&_mutex);
std::cout<<message<<gsep;
}
private:
MutexModule::Mutex _mutex;
};
//派生类: 往文件中打印日志
const std::string defaultpath = "./log";
const std::string defaultfilename = "my.log";
class FileLogStrategy : public LogStrategy
{
public:
FileLogStrategy(const std::string& path = defaultpath,const std::string& filename = defaultfilename)
:_path(path)
,_filename(filename)
{
if(std::filesystem::exists(_path)) //路径对应的文件/文件夹是否存在
return;
try
{
std::filesystem::create_directories(_path);//创建路径
}
catch(const std::filesystem::filesystem_error& e)
{
std::cerr<<e.what()<<"\\n";
}
}
~FileLogStrategy()
{}
void SyncLog(const std::string& message) override
{
MutexModule::LockGuard lg(&_mutex);
std::string file = _path + (_path.back() == '/' ? "" : "/") + _filename;
std::ofstream out(file,std::ios::app); //每次追加式写入
if(!out.is_open())
{
std::cout<<"创建"<<file<<"失败"<<std::endl;
return;
}
out<<message<<gsep;
out.close();
}
private:
std::string _path;
std::string _filename;
MutexModule::Mutex _mutex;
};
//日志等级
enum class LogLevel //强类型枚举
{
DEBUG, //调试信息
INFO, //普通信息
WARMING,//警告
ERROR, //错误
FATAL //致命错误
};
std::string LevelToStr(LogLevel level) //日志等级转字符串
{
switch(level)
{
case LogLevel::DEBUG:
return "DEBUG";
case LogLevel::ERROR:
return "ERROR";
case LogLevel::FATAL:
return "FATAL";
case LogLevel::INFO:
return "INFO";
case LogLevel::WARMING:
return "WARMING";
default:
return "UNKNOW";
}
}
std::string GetTimeStamp() //获取时间
{
time_t curtime = time(nullptr);
struct tm curtm;
localtime_r(&curtime,&curtm);
char timebuffer[64];
snprintf(timebuffer,sizeof(timebuffer),"%4d-%02d-%02d %02d:%02d:%02d",
curtm.tm_year+1900,
curtm.tm_mon+1,
curtm.tm_mday,
curtm.tm_hour,
curtm.tm_min,
curtm.tm_sec
);
return timebuffer;
}
class Logger
{
private:
std::unique_ptr<LogStrategy> _fflush_strategy;
public:
Logger()
{
EnableConsoleStrategy();
}
void EnableConsoleStrategy()
{
_fflush_strategy = std::make_unique<ConsoleLogStrategy>();
}
void EnableFileStrategy()
{
_fflush_strategy = std::make_unique<FileLogStrategy>();
}
void EnableFileStrategy(const std::string& path,const std::string& filename)
{
_fflush_strategy = std::make_unique<FileLogStrategy>(path,filename);
}
~Logger()
{}
class LogMessage
{
public:
LogMessage(LogLevel& level,const std::string& srcname,int line,Logger& logger)
:_level(level)
,_srcname(srcname)
,_line_num(line)
,_logger(logger)
,_cur_time(GetTimeStamp())
,_pid(getpid())
{
std::stringstream ss;
ss<<"["<<_cur_time<<"] "
<<"["<<LevelToStr(_level)<<"] "
<<"["<<_pid<<"] "
<<"["<<_srcname<<"] "
<<"["<<_line_num<<"] "
<<"- ";
_loginfo = ss.str();
}
~LogMessage()
{
if(_logger._fflush_strategy) //不为空，则输出日志
_logger._fflush_strategy->SyncLog(_loginfo);
}
// LogMessage(…) << "hell world" << "XXXX" << 3.14 << 1234
template <class T>
LogMessage& operator<<(const T& info)
{
std::stringstream ss;
ss<<info;
_loginfo += ss.str();
return *this; //以便支持类似连续赋值
}
private:
std::string _cur_time;//打印日志的时间
LogLevel _level; //日志等级
pid_t _pid; //进程pid
std::string _srcname; //打印对应⽇志的⽂件名
int _line_num; //行号
std::string _loginfo; //完整的日志信息
Logger& _logger;
};
LogMessage operator()(LogLevel level,const std::string& srcname,int line)
{
return LogMessage(level,srcname,line,*this);
}
};
Logger logger;
//宏
#define LOG(level) logger(level,__FILE__,__LINE__)
#define ENABLE_CONSOLE_LOGSTRATEGY() logger.EnableConsoleStrategy()
#define ENABLE_FILE_LOGSTRATEGY() logger.EnableFileStrategy()
}
#endif
测试代码：
#include <iostream>
#include "Log.hpp"
int main()
{
LogModule::logger(LogModule::LogLevel::DEBUG,"Main.cc",6)<<"Hello World! "<<123<<" "<<3.14159<<'\\n';
LogModule::logger.EnableFileStrategy();
LogModule::logger(LogModule::LogLevel::DEBUG,"Main.cc",6)<<"Hello World! "<<123<<" "<<3.14159<<" "<<"哈";
return 0;
}
#include <iostream>
#include "Log.hpp"
using namespace LogModule;
int main()
{
ENABLE_CONSOLE_LOGSTRATEGY();
LOG(LogLevel::INFO)<<"Hello Log! "<<1234<<" 呵呵";
ENABLE_FILE_LOGSTRATEGY();
LOG(LogLevel::WARMING)<<"Hello World! "<<3.14<<" 哈哈";
return 0;
}
线程池设计
底层向 OS 申请内存是系统调用，有时间成本。如果提前预分配一批资源缓存起来，就能减少系统调用次数，提升效率。线程池本质是池化技术：提前创建一批线程复用，避免频繁创建销毁线程带来系统调用开销。
线程池是一种线程复用的编程模式。线程过多会带来调度开销，破坏缓存局部性，降低整体性能。线程池预先维护一批线程，等待分配任务，避免频繁创建销毁线程的开销。
线程池的优势：充分利用 CPU 内核，防止过度调度；线程数量参考 CPU 核数、内存、网络 socket 数量。
线程池适合：
线程池不适合长时间运行的任务，例如 Telnet 长连接会话，任务运行时间远大于线程创建开销，线程池收益很低。
线程池分两类，我们实现固定数量线程池。
线程池的工作流程：用户把任务送入任务队列；线程池中的工作线程从队列取出任务，执行任务。
使用我们前面自己封装的 Thread 、Mutex 、Cond 、Log ，封装线程池。
成员变量如下：
std::vector<Thread> _threads; // 存放所有工作线程
int _num; // 线程池线程总数
std::queue<T> _taskq; // 任务队列，T是任务对象
Cond _cond; // 条件变量，线程休眠/唤醒
Mutex _mutex; // 互斥锁，保护任务队列
bool _isrunning; // 线程池运行标记，true运行；false停止
int _sleepernum; // 当前休眠等待任务的线程数量
我们希望创建 ThreadPool 对象后，调用 Start() 让线程池正式运行，调用 Stop() 终止线程池的工作，最后调用 Join() 回收线程。
HandlerTask() 经过lambda表达式包装后，作为 Thread 的构造函数 fun_t 参数传入。PushTask() 往任务队列中添加任务。
线程池的成员变量 _sleepernum 表示当前休眠等待任务的线程数量，它是共享资源，使用到的地方要加锁。
HandlerTask() 要使得任务队列为空但线程池还在运行时，调用 _cond.Wait() 阻塞线程。当有任务添加进来后，执行任务。
//Thread.hpp
#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <unistd.h>
#include <pthread.h>
#include <cstdint>
#include <cstring>
namespace ThreadModule
{
static uint32_t num = 1;
class Thread
{
using func_t = std::function<void()>;
void EnableDetach()
{
//std::cout<<_name<<"分离"<<std::endl;
_isDetach = true;
}
void EnableRunning()
{
_isRunning = true;
}
static void* Routine(void* arg) //不能使用成员函数，会多一个this指针
{
Thread* self = static_cast<Thread*>(arg);
self->EnableRunning(); //运行线程
if(self->_isDetach) //调用Detach前未Start
self->Detach();
//线程名放到线程局部存储，同时避免其他线程访问
pthread_setname_np(self->_tid,self->_name.c_str());
self->_func();
return nullptr;
}
public:
Thread(func_t func)
:_tid(0)
,_isDetach(false)
,_isRunning(false)
,_res(nullptr)
,_func(func)
{
_name = "thread – " + std::to_string(num++);
}
~Thread()
{}
void Detach()
{
if(_isDetach)
return;
if(_isRunning)
pthread_detach(_tid);
EnableDetach();
}
bool Start()
{
if(_isRunning)
return false;
int n = pthread_create(&_tid,nullptr,Routine,this);
if(n != 0)
{
return false;
}
else
{
return true;
}
}
bool Stop()
{
if(!_isRunning)
return false;
int n = pthread_cancel(_tid);
if(n != 0)
{
return false;
}
else
{
_isRunning = false;
return true;
}
}
bool Join()
{
if(_isDetach)
{
return false;
}
int n = pthread_join(_tid,&_res);
if(n != 0)
{
return false;
}
else
{
return true;
}
}
std::string GetName() {return _name;}
private:
pthread_t _tid;
std::string _name;
bool _isDetach;
bool _isRunning;
void* _res;
func_t _func;
};
}
//ThreadPool.hpp
#pragma once
#include <vector>
#include <queue>
#include <string>
#include "Mutex.hpp"
#include "Cond.hpp"
#include "Thread.hpp"
#include "Log.hpp"
namespace ThreadPoolModule
{
static const int gnum = 5;
template <class T>
class ThreadPool
{
void WakeUpOne()
{
_cond.Signal();
LogModule::LOG(LogModule::LogLevel::INFO)<<"唤醒一个线程";
}
void WakeUpAll()
{
MutexModule::LockGuard lg(&_mutex);
if(_sleepernum)
_cond.Broadcast();
LogModule::LOG(LogModule::LogLevel::INFO)<<"唤醒所有线程";
}
public:
ThreadPool(int num = gnum)
:_num(num)
,_sleepernum(0)
,_isrunning(false)
{
for(int i=0;i<_num;i++) //HandlerTask是非静态成员函数，有this指针，不能直接插入
{
_threads.emplace_back(
[this]()
{
HandlerTask();
}
);
}
}
~ThreadPool()
{}
void Start()
{
if(_isrunning)
return;
_isrunning = true;
for(ThreadModule::Thread& t:_threads)
{
if(t.Start())
LogModule::LOG(LogModule::LogLevel::INFO)<<t.GetName()<<" Start() success";
else
LogModule::LOG(LogModule::LogLevel::ERROR)<<"Start() failed";
}
}
void Stop()
{
if(!_isrunning)
return;
_isrunning = false;
WakeUpAll(); //每个线程不调用t.Stop，不暴力终止
}
void Join()
{
if(_isrunning)
return;
for(ThreadModule::Thread& t:_threads)
{
if(t.Join())
LogModule::LOG(LogModule::LogLevel::INFO)<<t.GetName()<<" Join() success";
else
LogModule::LOG(LogModule::LogLevel::ERROR)<<"Join() failed";
}
}
void HandlerTask()
{
char name[64];
/*类Thread中，Routine没有this指针，HandlerTask是线程池成员函数，
不知道调用它的是哪一个线程*/
pthread_getname_np(pthread_self(),name,sizeof(name));
while(1)
{
T t;
{
MutexModule::LockGuard lg(&_mutex);
while(_taskq.empty() && _isrunning)
{
_sleepernum++;
_cond.Wait(&_mutex);
_sleepernum—;
}
if(!_isrunning && _taskq.empty())
{
LogModule::LOG(LogModule::LogLevel::INFO)<<name<<"线程池停止工作且任务队列为空";
break;
}
t = _taskq.front();
_taskq.pop();
}
t();
}
}
bool PushTask(const T& in)
{
if(_isrunning)
{
MutexModule::LockGuard lg(&_mutex);
_taskq.push(in);
if(_num == _sleepernum)
WakeUpOne();
return true;
}
return false;
}
private:
int _num; //创建线程数目
std::vector<ThreadModule::Thread> _threads;
std::queue<T> _taskq; //任务队列
int _sleepernum; //休眠线程个数
bool _isrunning; //线程池是否正在运行
MutexModule::Mutex _mutex;
CondModule::Cond _cond;
};
}
测试代码：
//Task.hpp
#pragma once
#include <iostream>
#include <unistd.h>
#include <functional>
using fun_t = std::function<void()>;
void DownLoad()
{
LogModule::LOG(LogModule::LogLevel::INFO)<<"我是一个下载任务…";
sleep(3);
}
//Main.cc
#include <iostream>
#include <unistd.h>
#include "Task.hpp"
#include "ThreadPool.hpp"
using namespace ThreadPoolModule;
int main()
{
ThreadPool<fun_t> tp;
tp.Start();
sleep(2);
tp.PushTask(DownLoad);
tp.PushTask(DownLoad);
sleep(2);
tp.Stop();
tp.Join();
return 0;
}
单例模式
我们有时希望一个类在全局只能存在唯一一个实例对象。在服务器场景中，加载超大内存资源，就用单例对象统一管理这份资源。
单例模式的实现又分饿汉方式和懒汉方式。
饿汉方式：程序启动直接创建实例。吃饭立刻洗碗；对象在程序启动就构造完成；天然线程安全；但会拖慢启动速度。
懒汉方式：延时加载，第一次调用 GetInstance 才 new 对象。吃饭先放碗，要用的时候才洗碗；加快程序启动；多线程环境下原生代码线程不安全。
饿汉方式实现单例模式：
template <typename T>
class Singleton
{
static T data;
public:
static T* GetInstance()
{
return &data;
}
};
通过 Singleton 这个包装类来使用 T 对象，则一个进程中只有一个 T 对象的实例。
懒汉方式实现单例模式：
template <typename T>
class Singleton
{
static T* inst;
public:
static T* GetInstance()
{
if (inst == NULL)
{
inst = new T();
}
return inst;
}
};
如果已经创建了对象实例，则 inst != NULL ，直接返回。
懒汉方式实现单例模式（线程安全版本）：
template <typename T>
class Singleton
{
volatile static T* inst; // volatile禁止编译器过度优化
static std::mutex lock;
public:
static T* GetInstance()
{
// 外层判断：已经创建实例，直接跳过锁，提升性能
if(inst == nullptr)
{
lock.lock();
// 内层判断：防止多线程同时进入外层if，重复new
if(inst == nullptr)
{
inst = new T();
}
lock.unlock();
}
return inst;
}
};
封装单例模式的线程池。
GetInstance()为什么必须是 static？
- 单例的目标：对象还没有创建出来的时候，就要调用接口去创建对象。
- 静态成员函数不属于对象，属于类本身，不需要实例就可以调用。
保存实例的指针inc为什么是 static？
- 单例要求全局范围内只保存那唯一一份对象的地址，不属于任何一个实例，属于类。
- 静态成员变量存放在全局静态存储区。程序生命周期内只有一份，所有调用GetInstance()的地方共享这同一个指针，记录那唯一实例的地址。
锁_lock为什么是 static？
- 锁是用来保护多线程并发访问GetInstance()。
- 锁不能依附某个对象，对象还没创建，锁就要可用。所以锁必须是类的静态成员，全局只有一把。
构造私有化：禁止外部代码直接 ThreadPool<T> pool; 或者 new ThreadPool<T>() 手动创建对象。所有实例只能通过静态接口GetInstance() 获取唯一对象，保证全局只有一个线程池实例。
Start() 私有：线程池启动逻辑交给 GetInstance() 内部调用 inc->Start()。外部不能随意调用 Start() 重复启动线程池，防止外部乱调用导致多次创建工作线程。
单例模式逻辑上不允许拷贝或赋值，否则会破坏单例全局唯一的语义。
模板类里面写 static T var; 只是声明，不是定义。类内部 static xxx; 告诉编译器：有这么一个静态变量；但是没有分配内存。必须在类外部做模板静态成员的定义，编译器才会给这个静态变量分配存储空间。
//TP_Singleton.hpp
#pragma once
#include <vector>
#include <queue>
#include <string>
#include "Mutex.hpp"
#include "Cond.hpp"
#include "Thread.hpp"
#include "Log.hpp"
namespace ThreadPoolModule
{
static const int gnum = 5;
template <class T>
class ThreadPool
{
void WakeUpOne()
{
_cond.Signal();
LogModule::LOG(LogModule::LogLevel::INFO)<<"唤醒一个线程";
}
void WakeUpAll()
{
MutexModule::LockGuard lg(&_mutex);
if(_sleepernum)
_cond.Broadcast();
LogModule::LOG(LogModule::LogLevel::INFO)<<"唤醒所有线程";
}
ThreadPool(int num = gnum)
:_num(num)
,_sleepernum(0)
,_isrunning(false)
{
for(int i=0;i<_num;i++) //HandlerTask是非静态成员函数，有this指针，不能直接插入
{
_threads.emplace_back(
[this]()
{
HandlerTask();
}
);
}
}
void Start()
{
if(_isrunning)
return;
_isrunning = true;
for(ThreadModule::Thread& t:_threads)
{
if(t.Start())
LogModule::LOG(LogModule::LogLevel::INFO)<<t.GetName()<<" Start() success";
else
LogModule::LOG(LogModule::LogLevel::ERROR)<<"Start() failed";
}
}
ThreadPool(const ThreadPool<T>& tp) = delete;
ThreadPool<T>& operator=(const ThreadPool<T>& tp) = delete;
public:
~ThreadPool()
{}
static ThreadPool<T>* GetInstance()
{
if(_inc == nullptr)
{
MutexModule::LockGuard lg(&_lock);
if(_inc == nullptr)
{
LogModule::LOG(LogModule::LogLevel::INFO)<<"首次创建单例…";
_inc = new ThreadPool<T>();
_inc->Start();
}
}
return _inc;
}
void Stop()
{
if(!_isrunning)
return;
_isrunning = false;
WakeUpAll(); //每个线程不调用t.Stop，不暴力终止
}
void Join()
{
if(_isrunning)
return;
for(ThreadModule::Thread& t:_threads)
{
if(t.Join())
LogModule::LOG(LogModule::LogLevel::INFO)<<t.GetName()<<" Join() success";
else
LogModule::LOG(LogModule::LogLevel::ERROR)<<"Join() failed";
}
}
void HandlerTask()
{
char name[64];
/*类Thread中，Routine没有this指针，HandlerTask是线程池成员函数，
不知道调用它的是哪一个线程*/
pthread_getname_np(pthread_self(),name,sizeof(name));
while(1)
{
T t;
{
MutexModule::LockGuard lg(&_mutex);
while(_taskq.empty() && _isrunning)
{
_sleepernum++;
_cond.Wait(&_mutex);
_sleepernum—;
}
if(!_isrunning && _taskq.empty())
{
LogModule::LOG(LogModule::LogLevel::INFO)<<name<<"线程池停止工作且任务队列为空";
break;
}
t = _taskq.front();
_taskq.pop();
}
t();
}
}
bool PushTask(const T& in)
{
if(_isrunning)
{
MutexModule::LockGuard lg(&_mutex);
_taskq.push(in);
if(_num == _sleepernum)
WakeUpOne();
return true;
}
return false;
}
private:
int _num; //创建线程数目
std::vector<ThreadModule::Thread> _threads;
std::queue<T> _taskq; //任务队列
int _sleepernum; //休眠线程个数
bool _isrunning; //线程池是否正在运行
MutexModule::Mutex _mutex;
CondModule::Cond _cond;
static MutexModule::Mutex _lock;
static ThreadPool<T>* _inc;
};
template<class T>
MutexModule::Mutex ThreadPool<T>::_lock;
template<class T>
ThreadPool<T>* ThreadPool<T>::_inc = nullptr;
}
测试代码：
#include <iostream>
#include <unistd.h>
#include "Task.hpp"
#include "TP_Singleton.hpp"
using namespace ThreadPoolModule;
int main()
{
int cnt = 10;
while(cnt—)
{
ThreadPool<fun_t>::GetInstance()->PushTask(DownLoad);
sleep(1);
}
sleep(5);
ThreadPool<fun_t>::GetInstance()->Stop();
ThreadPool<fun_t>::GetInstance()->Join();
return 0;
}
线程安全和重入问题
线程安全：多个线程在访问共享资源时，能够正确地执行，不会相互干扰或破坏彼此的执行结果。
重入：同一个函数被不同的执行流调用，当前的一个流程还没有执行完，就有其他的执行流再次进入。
一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数。
两种重入场景：
常见的线程不安全的情况：
- 不保护共享变量的函数。
- 函数状态随着被调用，状态发生变化的函数。
- 返回指向静态变量指针的函数。
- 调用线程不安全函数的函数。
常见的线程安全的情况：
- 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的。
- 类或者接口对于线程来说都是原子操作。
- 多个线程之间的切换不会导致该接口的执行结果存在二义性。
常见不可重入的情况：
- 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的。
- 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构。
- 可重入函数体内使用了静态的数据结构。
常见可重入的情况：
- 不使用全局变量或静态变量。
- 不使用 malloc或者new开辟出的空间。
- 不调用不可重入函数。
- 不返回静态或全局数据，所有数据都有函数的调用者提供。
- 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据。
总结：
- 函数是可重入的，那就是线程安全的。
- 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题。
- 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。
- 可重入函数是线程安全函数的一种。线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
- 线程安全侧重说明线程访问公共资源的安全情况，表现的是并发线程的特点。
- 可重入描述的是一个函数是否能被重复进入，表示的是函数的特点。
死锁
死锁：一组线程，各自拿着已经占有的锁不释放，又互相请求对方手里的锁，所有线程永久阻塞，谁都无法继续往下执行。
典型场景：
- 线程 A：拿到锁 1，再去申请锁 2。
- 线程 B：拿到锁 2，再去申请锁 1。
A 等 B 释放锁 2，B 等 A 释放锁 1，互相等待，程序卡死。申请一把锁是原子的，不会死锁；同时申请多把锁，顺序错乱就容易死锁。
死锁的四个必要条件：
互斥条件
资源（锁）同一时刻只能给一个线程使用。互斥锁本身特性。
请求与保持（占有且等待）
线程已经占有一部分锁，不释放，又去申请新的锁。
不可剥夺条件
线程不能强行把已经被占有的锁抢走。只能持有者自己主动 unlock 释放。
循环等待条件
线程之间形成环形等待链。
避免死锁（破坏四大条件）：
破坏请求与保持：一次性申请所有锁
需要锁 1 + 锁 2，就一次性全部申请到手，不拿到全部就什么都不拿。C++ std::lock(mtx1,mtx2)，同时锁住多个互斥量，内部算法自动规避死锁。
破坏循环等待：统一加锁顺序
所有线程，永远按固定顺序拿锁，比如永远先拿 mtx1，再拿 mtx2。禁止 A：锁 1→锁 2；B：锁 2→锁 1 这种颠倒顺序。
破坏不可剥夺：锁拿不到，就把自己手上已经拿到的锁全部释放，过一会重试。
破坏互斥：能不用锁就不用锁（无锁编程，很少用）。
避免死锁的算法——银行家算法：操作系统预先判断资源分配之后会不会进入死锁状态，如果会就拒绝分配资源。不是发生死锁之后解决。银行家算法实际工程很少使用，更多是笔试考点。
STL、智能指针和线程安全
STL 容器默认不是线程安全，标准库不会内部帮你加锁。STL 追求极致性能，内置锁会带来不小性能损耗；使用场景千差万别：有的场景读多写少、有的场景需要细粒度锁（锁桶 / 锁链表），库不知道你业务需要哪种锁，把加锁交给使用者自己处理。
unique_ptr 独占所有权，不涉及多线程共享，没有线程安全问题。资源只能被这一个 unique_ptr 持有，不能拷贝，只能 move。不会多线程同时操作同一个对象。
shared_ptr 引用计数是线程安全的，标准库内部用原子操作 (CAS) 修改 ref 计数。但它托管的资源对象，库不做任何保护，并发读写对象需要用户自己上锁。
评论前必须登录！
注册