【Linux】29：线程池的实现

目录

一、线程池设计

二、代码

2.1 Mutex.hpp

2.2 Cond.hpp

2.3 Thread.hpp

2.4 Log.hpp

2.5 Task.hpp

2.6 threadpool.hpp

2.7 main.cc

三、线程安全的单例模式

3.1 什么是单例模式

3.2 单例模式的特点

3.3 饿汉实现方式和懒汉实现方式

3.4 饿汉实现单例模式

3.5 懒汉实现单例模式

3.6 懒汉实现单例模式（线程安全版本）

四、单例模式的线程池

4.1 threadpool.hpp

4.2 main.cc

五、线程安全和重入问题

5.1 概念

5.2 结论

六、常见的锁概念

6.1 死锁

6.2 死锁的四个必要条件

七、STL、智能指针、线程安全

7.1 STL容器是否是线程安全的

7.2 智能指针是否是线程安全的

一、线程池设计

线程池介绍：

一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

线程池的应用场景：

• 需要大量的线程来完成任务，且完成任务的时间比较短。比如WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
• 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
• 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.

线程池的种类：

• 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口
• 浮动线程池，其他同上

此处，我们选择固定线程个数的线程池。

二、代码

2.1 Mutex.hpp

#pragma once #include <iostream> #include <pthread.h> namespace MutexModule { class Mutex { public: Mutex() { pthread_mutex_init(&_mutex, nullptr); } void Lock() { pthread_mutex_lock(&_mutex); } void UnLock() { pthread_mutex_unlock(&_mutex); } pthread_mutex_t *Get() { return &_mutex; } ~Mutex() { pthread_mutex_destroy(&_mutex); } private: pthread_mutex_t _mutex; }; class LockGuard { public: LockGuard(Mutex &mutex) : _mutex(mutex) { _mutex.Lock(); } ~LockGuard() { _mutex.UnLock(); } private: Mutex &_mutex; }; }

2.2 Cond.hpp

#pragma once #include <iostream> #include <pthread.h> #include "Mutex.hpp" using namespace MutexModule; namespace CondModule { class Cond { public: Cond() { pthread_cond_init(&_cond, nullptr); } void Wait(Mutex &mutex) { pthread_cond_wait(&_cond, mutex.Get()); } void Singal() { pthread_cond_signal(&_cond); } void BroadCast() { pthread_cond_broadcast(&_cond); } ~Cond() { pthread_cond_destroy(&_cond); } private: pthread_cond_t _cond; }; }

2.3 Thread.hpp

#ifndef _THREAD_H_ #define _THREAD_H_ #include <iostream> #include <string> #include <cstdio> #include <unistd.h> #include <cstring> #include <pthread.h> #include <functional> #include "Log.hpp" using namespace LogModule; namespace ThreadModule { static uint32_t number = 1; class Thread { using func_t = std::function<void()>; void EnableDetach() { // std::cout << "线程被分离了" << std::endl; _isdetach = true; } public: Thread(func_t func) : _tid(0), _isdetach(false), _isrunning(false), _res(nullptr), _func(func) { _name = "thread-" + std::to_string(number++); } void Detach() { if (_isdetach) // 如果线程已经分离了，就不需要再分离了。 return; if (_isrunning) pthread_detach(_tid); EnableDetach(); } void EnableRunning() { _isrunning = true; } static void *Routine(void *args) { Thread *self = (Thread *)args; self->EnableRunning(); if (self->_isdetach) // 如果线程的分离标志是true，就调用系统调用pthread_detach来将线程设置分离 self->Detach(); pthread_setname_np(self->_tid, self->_name.c_str()); self->_func(); return nullptr; } bool Start() { if (_isrunning) return false; int n = pthread_create(&_tid, nullptr, Routine, this); if (n != 0) { // std::cerr << "create pthread error" << strerror(n) << std::endl; return false; } else { // std::cout << _name << " create sucess" << std::endl; } return true; } bool Stop() { if (_isrunning) { int n = pthread_cancel(_tid); if (n != 0) { // std::cerr << "cancel pthread error" << strerror(n) << std::endl; return false; } else { _isrunning = false; // std::cerr << _name << " stop" << std::endl; } } return true; } bool Join() { if (_isdetach) { // std::cout << "你的线程已经是分离的，不能进行join" << std::endl; return true; } int n = pthread_join(_tid, &_res); if (n != 0) { // std::cerr << "join pthread error" << strerror(n) << std::endl; LOG(LogLevel::DEBUG) << " join线程失败"; return false; } else { // std::cerr << "join pthread sucess" << std::endl; LOG(LogLevel::DEBUG) << " join线程成功"; } return true; } std::string GetName() { return _name; } ~Thread() {} private: pthread_t _tid; // 线程id std::string _name; // 线程名字 bool _isdetach; // 是否分离，默认不分离 bool _isrunning; // 线程状态，是否运行 void *_res; // 线程返回值 func_t _func; // 线程执行的任务 }; } #endif

2.4 Log.hpp

#ifndef __LOG_HPP #define __LOG_HPP #include <iostream> #include <filesystem> #include <string> #include <memory> #include <unistd.h> #include <fstream> #include <ctime> #include <cstdio> #include "Mutex.hpp" namespace LogModule { using namespace MutexModule; const std::string gsep = "\r\n"; // 刷新策略基类 class LogStrategy { public: ~LogStrategy() = default; virtual void SyncLog(const std::string &message) = 0; }; // 显示器打印日志的策略 class ConsoleLogStrategy : public LogStrategy { public: ConsoleLogStrategy() {} ~ConsoleLogStrategy() {} void SyncLog(const std::string &message) override { LockGuard lockguard(_mutex); std::cout << message << gsep; } private: Mutex _mutex; }; // 文件打印日志的策略 const std::string defaultpath = "./log"; const std::string defaultfile = "my.log"; class FileLogStrategy : public LogStrategy { public: FileLogStrategy(const std::string &path = defaultpath, const std::string &file = defaultfile) : _path(path), _file(file) { LockGuard lockguard(_mutex); // 来判断文件路径是否存在 if (std::filesystem::exists(_path)) { return; } try { std::filesystem::create_directories(_path); } catch (const std::filesystem::filesystem_error &e) { std::cerr << e.what() << "\n"; } } void SyncLog(const std::string &message) override { LockGuard lockguard(_mutex); std::string filename = _path + (_path.back() == '/' ? "" : "/") + _file; std::ofstream out(filename, std::ios::app); if (!out.is_open()) { return; } out << message << gsep; out.close(); } ~FileLogStrategy() {} private: std::string _path; // 日志文件所在路径 std::string _file; // 日志文件本身 Mutex _mutex; }; // 1：形成日志等级 enum class LogLevel { DEBUG, INFO, WARNING, ERROR, FATAL }; std::string Level2Str(LogLevel level) { switch (level) { case LogLevel::DEBUG: return "DEBUG"; case LogLevel::INFO: return "INFO"; case LogLevel::WARNING: return "WARNING"; case LogLevel::ERROR: return "ERROR"; case LogLevel::FATAL: return "FATAL"; default: return "UNKNOWN"; } } // 获得时间 std::string GetTimeStamp() { time_t curr = time(nullptr); struct tm curr_tm; localtime_r(&curr, &curr_tm); char timebuffer[128]; snprintf(timebuffer, sizeof(timebuffer), "%4d-%02d-%02d %02d:%02d:%02d", curr_tm.tm_year + 1900, curr_tm.tm_mon + 1, curr_tm.tm_mday, curr_tm.tm_hour, curr_tm.tm_min, curr_tm.tm_sec); return timebuffer; } // 1：形成日志 2根据不同的策略，完成刷新 class Logger { public: Logger() { EnableConsoleLogStrategy(); } void EnableFileLogStrategy() { _fflush_strategy = std::make_unique<FileLogStrategy>(); } void EnableConsoleLogStrategy() { _fflush_strategy = std::make_unique<ConsoleLogStrategy>(); } // 内部类，表示的是未来的一条日志 class logMessage { public: logMessage(LogLevel &level, std::string &src_name, int line_number, Logger &logger) : _curr_time(GetTimeStamp()), _level(level), _pid(getpid()), _src_name(src_name), _line_number(line_number), _logger(logger) { // 日志的左半部分拼进来 std::stringstream ss; ss << "[" << _curr_time << "]" << "[" << Level2Str(_level) << "]" << "[" << _pid << "]" << "[" << _src_name << "]" << "[" << _line_number << "]" << "-"; _loginfo = ss.str(); } template <typename T> logMessage &operator<<(const T &info) { std::stringstream ss; ss << info; _loginfo += ss.str(); return *this; } ~logMessage() { if (_logger._fflush_strategy) { _logger._fflush_strategy->SyncLog(_loginfo); } } private: std::string _curr_time; LogLevel _level; pid_t _pid; std::string _src_name; int _line_number; std::string _loginfo; // 合并之后一条完整的信息 Logger &_logger; }; logMessage operator()(LogLevel level, std::string name, int line) { return logMessage(level, name, line, *this); } ~Logger() {} private: std::unique_ptr<LogStrategy> _fflush_strategy; }; // 定义全局日志对象 Logger logger; // 使用宏，简化用户操作 #define LOG(level) logger(level, __FILE__, __LINE__) #define Enable_Console_Log_Strategy() logger.EnableConsoleLogStrategy() #define Enable_File_Log_Strategy() logger.EnableFileLogStrategy() } #endif

2.5 Task.hpp

#pragma once #include <iostream> #include <unistd.h> #include <functional> #include "Log.hpp" using namespace LogModule; using task_t = std::function<void()>; void DownLoad() { LOG(LogLevel::DEBUG) << "我是一个下载任务"; }

2.6 threadpool.hpp

#pragma once #include <iostream> #include <string> #include <vector> #include <queue> #include "Log.hpp" #include "Thread.hpp" #include "Mutex.hpp" #include "Cond.hpp" namespace ThreadPoolModule { using namespace LogModule; using namespace MutexModule; using namespace CondModule; using namespace ThreadModule; static const int gnum = 5; template <typename T> class ThreadPool { void WakeUpAllthread() { LockGuard lockguard(_mutex); if (_sleepnum > 0) _cond.BroadCast(); LOG(LogLevel::INFO) << "唤醒所有的线程"; } void WakeUpOne() { _cond.Singal(); LOG(LogLevel::INFO) << "唤醒一个休眠的线程"; } public: ThreadPool(int num = gnum) : _num(num), _isrunning(false), _sleepnum(0) { for (int i = 0; i < _num; i++) { _threads.emplace_back( [this]() { HandlerTask(); }); } } void Start() { if (_isrunning) return; _isrunning = true; for (auto &t : _threads) { t.Start(); LOG(LogLevel::INFO) << "start new thread success " << t.GetName(); } } void Stop() { if (!_isrunning) return; _isrunning = false; // 需要唤醒所有线程 WakeUpAllthread(); } void Join() { for (auto &thread : _threads) { thread.Join(); } } void HandlerTask() { char name[128]; pthread_getname_np(pthread_self(), name, sizeof(name)); while (true) { T t; { LockGuard lockguard(_mutex); while (_isrunning && _taskq.empty()) { ++_sleepnum; _cond.Wait(_mutex); --_sleepnum; } // 线程被唤醒 if (!_isrunning && _taskq.empty()) { LOG(LogLevel::INFO) << name << " 退出了，线程池退出&&任务队列为空"; break; } // 一定有任务 t = _taskq.front(); _taskq.pop(); } t(); // 处理任务 } } bool Enqueue(const T &in) { if (_isrunning) { LockGuard lockguard(_mutex); _taskq.push(in); if (_threads.size() - _sleepnum == 0) WakeUpOne(); return true; } return false; } ~ThreadPool() {} private: std::vector<Thread> _threads; int _num; // 线程池中的线程个数 std::queue<T> _taskq; Cond _cond; Mutex _mutex; bool _isrunning; int _sleepnum; }; }

2.7 main.cc

#include "ThreadPool.hpp" #include "Task.hpp" #include "Log.hpp" using namespace ThreadPoolModule; int main() { Enable_Console_Log_Strategy(); ThreadPool<task_t> *tp = new ThreadPool<task_t>(); tp->Start(); int cnt = 10; while (cnt--) { tp->Enqueue(DownLoad); sleep(1); } tp->Stop(); tp->Join(); return 0; }

三、线程安全的单例模式

3.1 什么是单例模式

单例模式属于创建型设计模式，核心目标：保证一个类在整个程序中，有且仅有一个实例对象，并且提供一个全局访问点来获取这个实例。

简单一句话：一个类，只能 new 出来一个对象，大家共用这同一个对象。

3.2 单例模式的特点

某些类，只应该具有一个对象(实例)，就称之为单例。

例如一个男人只能有一个媳妇，

在很多服务器开发场景中，经常需要让服务器加载很多的数据（上百G）到内存中，此时往往要用一个单例的类来管理这些数据.

3.3 饿汉实现方式和懒汉实现方式

[洗碗的例子]
1：吃完饭，立刻洗碗，这种就是饿汉方式，因为下一顿吃的时候可以立刻拿着碗就能吃饭.
2：吃完饭，先把碗放下，然后下一顿饭用到这个碗了再洗碗，就是懒汉方式.

懒汉方式最核心的思想是"延时加载".从而能够优化服务器的启动速度。

3.4 饿汉实现单例模式

程序启动就创建，线程安全

template <typename T> class Singleton { static T data; public: static T* GetInstance() { return &data; } };

C++11 及以后，局部静态变量是线程安全（最推荐写法，又叫 Meyers Singleton）

3.5 懒汉实现单例模式

第一次调用才创建，延迟加载，要注意多线程

template <typename T> class Singleton { static T* inst; public: static T* GetInstance() { if (inst == NULL) { inst = new T(); } return inst; } };

原生懒汉式多线程下不安全，需要加锁（双重检查锁 DCL）

存在一个严重的问题，线程不安全，

第一次调用GetInstance的时候，如果两个线程同时调用，可能会创建出两份T对象的实例.

但是后续再次调用，就没有问题了.

3.6 懒汉实现单例模式（线程安全版本）

// 懒汉模式, 线程安全 template <typename T> class Singleton { volatile static T* inst; // 需要设置 volatile 关键字, 否则可能被编译器优化. static std::mutex lock; public: static T* GetInstance() { if (inst == NULL) { // 双重判定空指针, 降低锁冲突的概率, 提⾼性能. lock.lock(); // 使⽤互斥锁, 保证多线程情况下也只调⽤⼀次 new. if (inst == NULL) { inst = new T(); } lock.unlock(); } return inst; } };

四、单例模式的线程池

4.1 threadpool.hpp

#pragma once #include <iostream> #include <string> #include <vector> #include <queue> #include "Log.hpp" #include "Thread.hpp" #include "Mutex.hpp" #include "Cond.hpp" namespace ThreadPoolModule { using namespace LogModule; using namespace MutexModule; using namespace CondModule; using namespace ThreadModule; static const int gnum = 5; template <typename T> class ThreadPool { void WakeUpAllthread() { LockGuard lockguard(_mutex); if (_sleepnum > 0) _cond.BroadCast(); LOG(LogLevel::INFO) << "唤醒所有的线程"; } void WakeUpOne() { _cond.Singal(); LOG(LogLevel::INFO) << "唤醒一个休眠的线程"; } ThreadPool(int num = gnum) : _num(num), _isrunning(false), _sleepnum(0) { for (int i = 0; i < _num; i++) { _threads.emplace_back( [this]() { HandlerTask(); }); } } ThreadPool(const ThreadPool<T> &) = default; ThreadPool<T> &operator=(const ThreadPool<T> &) = default; void Start() { if (_isrunning) return; _isrunning = true; for (auto &t : _threads) { t.Start(); LOG(LogLevel::INFO) << "start new thread success " << t.GetName(); } } public: static ThreadPool<T> *GetInstance() { if (inc == nullptr) { LockGuard lockguard(_lock); LOG(LogLevel::DEBUG) << "获取单例"; if (inc == nullptr) { LOG(LogLevel::DEBUG) << "首次使用单例,创建线程池对象"; inc = new ThreadPool<T>(); inc->Start(); } } return inc; } void Stop() { if (!_isrunning) return; _isrunning = false; // 需要唤醒所有线程 WakeUpAllthread(); } void Join() { for (auto &thread : _threads) { thread.Join(); } } void HandlerTask() { char name[128]; pthread_getname_np(pthread_self(), name, sizeof(name)); while (true) { T t; { LockGuard lockguard(_mutex); while (_isrunning && _taskq.empty()) { ++_sleepnum; _cond.Wait(_mutex); --_sleepnum; } // 线程被唤醒 if (!_isrunning && _taskq.empty()) { LOG(LogLevel::INFO) << name << " 退出了，线程池退出&&任务队列为空"; break; } // 一定有任务 t = _taskq.front(); _taskq.pop(); } t(); // 处理任务 } } bool Enqueue(const T &in) { if (_isrunning) { LockGuard lockguard(_mutex); _taskq.push(in); if (_threads.size() - _sleepnum == 0) WakeUpOne(); return true; } return false; } ~ThreadPool() {} private: std::vector<Thread> _threads; int _num; // 线程池中的线程个数 std::queue<T> _taskq; Cond _cond; Mutex _mutex; bool _isrunning; int _sleepnum; static ThreadPool<T> *inc; // 单例指针 static Mutex _lock; }; template <typename T> ThreadPool<T> *ThreadPool<T>::inc = nullptr; template <typename T> Mutex ThreadPool<T>::_lock; }

4.2 main.cc

#include "ThreadPool.hpp" #include "Task.hpp" #include "Log.hpp" using namespace ThreadPoolModule; int main() { Enable_Console_Log_Strategy(); // ThreadPool<task_t> *tp = new ThreadPool<task_t>(); // tp->Start(); int cnt = 10; while (cnt--) { ThreadPool<task_t>::GetInstance()->Enqueue(DownLoad); sleep(1); } ThreadPool<task_t>::GetInstance()->Stop(); ThreadPool<task_t>::GetInstance()->Join(); return 0; }

五、线程安全和重入问题

5.1 概念

线程安全：就是多个线程在访问共享资源时，能够正确地执行，不会相互干扰或破坏彼此的执行结果。一般而言，多个线程并发同一段只有局部变量的代码时，不会出现不同的结果。但是对全局变量或者静态变量进行操作，并且没有锁保护的情况下，容易出现该问题。

重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数。

学到现在，其实我们已经能理解重入其实可以分为两种情况

• 多线程重入函数
• 信号导致一个执行流重复进入函数

常见的线程不安全的情况

• 不保护共享变量的函数
• 函数状态随着被调用，状态发生变化的函数
• 返回指向静态变量指针的函数
• 调用线程不安全函数的函数

常见的线程安全的情况

• 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的类或者接口对于线程来说都是原子操作
• 多个线程之间的切换不会导致该接口的执行
• 结果存在二义性

常见不可重入的情况

• 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
• 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
• 可重入函数体内使用了静态的数据结构

常见可重入的情况

• 不使用全局变量或静态变量
• 不使用用malloc或者new开辟出的空间
• 不调用不可重入函数
• 不返回静态或全局数据，所有数据都有函数的调用者提供
• 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

5.2 结论

不要被上面绕口令式的话语唬住，你只要仔细观察，其实对应概念说的都是一回事。

可重入与线程安全联系

• 函数是可重入的，那就是线程安全的（其实知道这一句话就够了）
• 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
• 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。

可重入与线程安全区别

• 可重入函数是线程安全函数的一种
• 线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
• 如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的。

注意：

• 如果不考虑信号导致一个执行流重复进入函数这种重入情况，线程安全和重入在安全角度不做区分
• 但是线程安全侧重说明线程访问公共资源的安全情况，表现的是并发线程的特点
• 可重入描述的是一个函数是否能被重复进入，表示的是函数的特点

六、常见的锁概念

6.1 死锁

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所占用不会释放的资源而处于的一种永久等待状态。

为了方便表述，假设现在线程A，线程B必须同时持有锁1和锁2，才能进行后续资源的访问

申请一把锁是原子的，但是申请两把锁就不一定了

造成的结果是：

6.2 死锁的四个必要条件

1. 互斥条件：一个资源每次只能被一个执行流使用
2. 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
3. 不剥夺条件：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
4. 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系

七、STL、智能指针、线程安全

7.1 STL容器是否是线程安全的

不是.

原因是，STL的设计初衷是将性能挖掘到极致，而一旦涉及到加锁保证线程安全，会对性能造成巨大的影响.

而且对于不同的容器，加锁方式的不同，性能可能也不同(例如hash表的锁表和锁桶)。

因此STL默认不是线程安全，如果需要在多线程环境下使用，往往需要调用者自行保证线程安全，

7.2 智能指针是否是线程安全的

对于unique_ptr，由于只是在当前代码块范围内生效，因此不涉及线程安全问题.

对于shared_ptr，多个对象需要共用一个引用计数变量，所以会存在线程安全问题.但是标准库实现的时候考虑到了这个问题，基于原子操作(CAS)的方式保证shared_ptr能够高效，原子的操作引用计数.