Java 中有一个很方便的 ThreadPoolExecutor,可以用做线程池。想找一下 C++ 的类似设施,尤其是能方便理解底层原理可上手的。网上找到的 demo,基本都是介绍的 projschj 的C++11线程池。这份源码最后的commit日期是2014年,现在是2021年了,本文将在阅读源码的基础上,对这份代码进行一些改造。关于线程池,目前网上讲解最好的一篇文章是这篇 Java线程池实现原理及其在美团业务中的实践,值得一读。
改造后的源码在 https://gitee.com/zhcpku/ThreadPool 进行提供。
主要包含两个部分,一组执行线程、一个任务队列。执行线程空闲时,总是从任务队列中取出任务执行。具体执行逻辑后面会进行解释。
class ThreadPool { // ... private: using task_type = std::function<void()>; // need to keep track of threads so we can join them std::vector<std::thread> workers; // the task queue std::queue<task_type> tasks; };
这里包括一把锁、一个条件变量,还有一个bool变量:
class ThreadPool { // ... private: // synchronization std::mutex queue_mutex; std::condition_variable condition; bool stop; };
启动线程池,首先要做的是构造指定数量的线程出来,然后让每个线程开始运行。
对于每个线程,运行逻辑是一样的:尝试从任务队列中获取任务并执行,如果拿不到任务、并且线程池没有被停用,则睡眠等待。
这里线程等待任务使用的是条件变量,而不是信号量或者自旋锁等其他设施,是为了让线程睡眠,避免CPU空转浪费。
// the constructor just launches some amount of workers inline ThreadPool::ThreadPool(size_t thread_num) : stop(false) { for (size_t i = 0; i < thread_num; ++i) { workers.emplace_back([this] { for (;;) { task_type task; { std::unique_lock<std::mutex> lock(this->queue_mutex); this->condition.wait( lock, [this] { return this->stop || !this->tasks.empty(); }); if (this->stop && this->tasks.empty()) { return; } task = std::move(this->tasks.front()); this->tasks.pop(); } task(); } }); } }
线程的停用,需要让每一个线程停下来,并且等到每个线程都停止再退出主线程才是比较安全的操作。
停止分三步:设置停止标识、通知到每一个线程(睡眠的线程需要唤醒)、等到每一个线程停止。
// the destructor joins all threads inline ThreadPool::~ThreadPool() { { std::unique_lock<std::mutex> lock(queue_mutex); stop = true; } condition.notify_all(); for (std::thread& worker : workers) { worker.join(); } }
这是整个线程池的核心,也是写的最复杂,用C++新特性最多的地方,包括但不限于:
自动类型推导、变长模板函数、右值引用、完美转发、原地构造、智能指针、future、bind ……
顺带提一句,要是早有变长模板参数,std::min / std::max 也不至于只能比较两个数大小,再多就得用大括号包起来作为 initialize_list 传进去了。
这里提交任务时,由于我们的任务类型定义为一个无参无返回值的函数对象,所以需要先通过 std::bind 把函数及其参数打包成一个 对应类型的可调用对象,返回值将通过 future 异步获取。然后是要把这个任务插入任务队列末尾,因为任务队列被多线程并发访问,所以需要加锁。
另外需要处理的两个情况,一个是线程睡眠时,新入队任务需要主要唤醒线程;另一个是线程池要停用时,入队操作是非法的。
// add new work item to the pool template <class F, class... Args> auto ThreadPool::enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> { using return_type = typename std::result_of<F(Args...)>::type; auto task = std::make_shared<std::packaged_task<return_type()>>( std::bind(std::forward<F>(f), std::forward<Args>(args)...)); std::future<return_type> res = task->get_future(); { std::unique_lock<std::mutex> lock(queue_mutex); // don't allow enqueueing after stopping the pool if (stop) { throw std::runtime_error("enqueue on stopped ThreadPool"); } tasks.emplace([task]() { (*task)(); }); } condition.notify_one(); return res; }
以上代码已经足以阐释线程池基本原理了,以下改进主要从可靠性、易用性、使用场景等方面进行改进。
线程池本身应该是不可复制的,这里我们通过删除拷贝构造函数和赋值操作符,以及其对用的右值引用版本来实现:
class ThreadPool { // ... private: // non-copyable ThreadPool(const ThreadPool&) = delete; ThreadPool(ThreadPool&&) = delete; ThreadPool& operator=(const ThreadPool&) = delete; ThreadPool& operator=(ThreadPool&&) = delete; };
除了手动指定线程个数,更合适的做法是主动探测CPU支持的物理线程数,并以此作为执行线程个数:
class ThreadPool { public: explicit ThreadPool(size_t thread_num = std::thread::hardware_concurrency()); size_t ThreadCount() { return workers.size(); } // ... };
线程不必一次就创建出来,可以等到任务到来的时候再创建,降低资源占用。
// TBD
线程池的应用场景主要针对的是CPU密集型应用,但是遇到IO密集型场景,也要保证可用性。如果我们的线程个数固定的话,会出现一些问题,比如:
上面的线程池,其启动停止时机分别是构造和析构的时候,还是太粗糙了。我们为其提供手动启动、停止的函数,并支持停止之后重新启动:
// TBD
不干了,2021年了,研究协程库去了!
参考文献