C++ 线程池设计与实现
线程池概述
线程池是一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络 sockets 等的数量。
应用场景
- 需要大量的线程来完成任务，且完成任务的时间比较短。例如 WEB 服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。但对于长时间的任务，比如一个 Telnet 连接请求，线程池的优点就不明显了。因为 Telnet 会话时间比线程的创建时间大多了。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的海量请求，但不至于使服务器因此产生大量线程的应用。突发性海量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。
种类
- 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口。
- 浮动线程池，其他同上。
线程描述与组织
既然是一个线程池，需要创建大量的线程，让线程从任务队列中取数据来执行任务。要对线程做创建，控制，回收，因此需要对线程做管理，如何管理呢？需要能够描述线程和做组织！
好在我们之前做过了对线程的封装，其封装如下：
// Thread.hpp
#define get_lwp_id() syscall(SYS_gettid)
using func_t = std::function<void(const std::string&name)>;
const std::string threadnamedefault = "None-Name";
class Thread {
public:
Thread(func_t func, const std::string &name = threadnamedefault)
: _name(name), _func(func), _isrunning(false) {
LOG(LogLevel::INFO) << _name << " create thread obj success";
}
static void *start_routine(void *args) {
Thread *self = static_cast<Thread *>(args);
self->_isrunning = true;
self->_lwpid = get_lwp_id();
self->_func(self->_name);
pthread_exit((void *)0);
}
void Start() {
int n = pthread_create(&_tid, nullptr, start_routine, this);
if (n == 0) {
LOG(LogLevel::INFO) << _name << " running success";
}
}
void Stop() {
int n = pthread_cancel(_tid); // 太简单粗暴了
(void)n;
}
// 检测线程结束并且回收的功能
void Join() {
if (!_isrunning) return;
int n = pthread_join(_tid, nullptr);
if (n == 0) {
LOG(LogLevel::INFO) << _name << " pthread_join success";
}
}
~Thread() {
// LOG(LogLevel::INFO) << _name << " destory thread obj success";
}
private:
bool _isrunning;
pthread_t _tid;
pid_t _lwpid;
std::string _name;
func_t _func;
};
#endif
线程管理
我们使用 _q 队列来表示任务队列，既然存在很多线程从任务队列里取数据，那这不就是多线程并发访问共享资源吗？因此，需要对共享资源做保护啊！如何保护呢？互斥锁。可是，该加一把锁呢还是，这里以一把锁就可以解决多线程并发访问共享资源的问题。
#pragma once
#include <pthread.h>
#include <iostream>
#include <unistd.h>
#include <string>
class Mutex {
public:
Mutex() {
// 初始化锁
pthread_mutex_init(&_lock, nullptr);
}
void Lock() {
// 加锁
pthread_mutex_lock(&_lock);
}
void Unlock() {
// 解锁
pthread_mutex_unlock(&_lock);
}
pthread_mutex_t *Get() { return &_lock; }
~Mutex() {
// 毁坏锁
pthread_mutex_destroy(&_lock);
}
private:
pthread_mutex_t _lock;
};
class LockGuard {
public:
LockGuard(Mutex *mutex) : _mutexp(mutex) {
_mutexp->Lock();
}
~LockGuard() {
_mutexp->Unlock();
}
private:
Mutex *_mutexp;
};
可是，单单只有锁是不够的，会存在这种问题。如果某个线程的竞争能力特别强呢？我用户都没往任务队列里放数据，你这个线程就不断加锁，检查任务队列是否有数据，在解锁，这反而会导致效率降低。那么能不能当用户往任务队列里放数据后，再通知线程们去执行这个任务呢？当然可以，如果没有放数据，此时线程们都需要进入休眠，需要使用到条件变量。
#pragma once
#include<pthread.h>
#include"Mutex.hpp"
class Cond {
public:
Cond() {
pthread_cond_init(&_cond,nullptr);
}
void Wait(Mutex &mutex) {
int n = pthread_cond_wait(&_cond,mutex.Get());
}
void NotifyOne() {
int n = pthread_cond_signal(&_cond);
(void)n;
}
void NotifyAll() {
int n = pthread_cond_broadcast(&_cond);
(void)n;
}
~Cond() {
pthread_cond_destroy(&_cond);
}
private:
pthread_cond_t _cond;
};
线程池实现
前置工作准备完毕，那么就可以开始对线程池的封装了！
template <class T> class ThreadPool {
private:
// 任务队列
std::queue<T> _q;
// 整体使用的临界资源
// 多个线程
std::vector<Thread> _threads;
int _threadnum;
int _wait_thread_num;
// 保护机制
Mutex _lock;
Cond _cond;
// 其他属性
bool _is_running;
};
这里的 std::vector<Thread> _threads 是否创建了线程对象呢？可以肯定的是肯定是没有的，相当于你只是拥有了线程这个图纸，里面含有线程的相关属性。
构造函数
// 线程池
const static int defaultthreadnum = 3; // for debug
template <class T> class ThreadPool {
private:
void Routine(const std::string &name) {
while (true) {
// 把任务从线程获取到线程私有！临界区 -> 私有的栈
T t;
{
LockGuard lockguard(&_lock);
while (QueueIsEmpty() && _is_running) {
_wait_thread_num++;
_cond.Wait(_lock);
_wait_thread_num--;
}
if (!_is_running && QueueIsEmpty()) {
LOG(LogLevel::INFO) << " 线程池退出 && 任务队列为空，" << name << " 退出";
break;
}
// 队列中一定有任务了!, 但是
// 1. 线程池退出 -- 消耗历史
// 2. 线程池没有退出 -- 正常工作
t = _q.front();
_q.pop();
}
t(); // 规定，未来的任务，必须这样处理！
// for debug
// LOG(LogLevel::DEBUG) << name << " handler task: " << t.Result2String();
}
}
public:
ThreadPool(int threadnum = defaultthreadnum)
: _threadnum(threadnum), _is_running(false), _wait_thread_num(0) {
for (int i = 0; i < _threadnum; i++) {
std::string name = "thread-" + std::to_string(i + 1);
_threads.emplace_back([this](const std::string &name) {
this->Routine(name);
}, name);
}
LOG(LogLevel::INFO) << "thread pool obj create success";
}
};
问题 1: 构造函数这里的 for 循环时对象存在了吗？ 存在了的，在初始化列表这里对对象进行初始化，进行了开辟空间，已然可以访问类内属性。
问题 2: 构造函数并不需要启动线程池，只负责获取资源并设置其要执行的任务。只有当 ThreadPool 对象调用 Start 成员函数时，才需要让线程池启动起来。
我们的想法是这样的：在这里将 this 指针与任务执行方法进行绑定，当线程池启动起来时，能够调用 Thread 线程对象的 start 函数，里面的 pthread_create 函数会调用 start_routine 函数，函数体内的 self->_func(self->_name) 会因为包装器的原因，回调到 ThreadPool 的任务执行方法。这种思想可以做到模块与模块之间的解耦合。
线程启动、取消与回收
线程启动
void Start() {
if (_is_running) return;
_is_running = true;
for (auto &t : _threads) {
t.Start();
}
LOG(LogLevel::INFO) << "thread pool running success";
}
线程取消
void Stop() {
if (!_is_running) return;
_is_running = false;
if (_wait_thread_num) _cond.NotifyAll();
LOG(LogLevel::INFO) << "thread pool stop success";
}
但实际上上面这种做法并不推荐，太过于简单粗暴了，考虑的因素太少了。
一个线程池要退出时，应该让线程走征程的唤醒逻辑以及退出啊！
- 如果被唤醒 && 任务队列没有任务 = 让线程退出
- 如果被唤醒 && 任务队列有任务 = 线程不能立即退出，而应该让线程把任务处理完，在退出
- 线程本身没有被休眠，我们应该让他把他能处理的任务全部处理完成，在退出
可以发现第 3 种情况可以归到第 2 种情况中，线程取消如下所示，为什么是这样呢？我们来看看任务执行是怎样做的。
回收线程
void Wait() {
for (auto &t : _threads) {
t.Join();
}
LOG(LogLevel::INFO) << "thread pool wait success";
}
往线程池入数据
void Enqueue(const T &t) {
if (!_is_running) return;
{
LockGuard lockguard(&_lock);
_q.push(t);
if (_wait_thread_num > 0) _cond.NotifyOne();
}
}