简单的线程池（二）

日期：2022-03-23 栏目：程序人生浏览：次

作者在简单的线程池中采用了非阻塞的（nonblocking）线程同步方式，在此文中作者将采用阻塞的（blocking）线程同步方式实现相同特性的线程池。

本文中不再赘述与简单的线程池相同的内容。如有不明之处，请参考该博客。

实现

以下代码给出了此线程池的实现。

class Thread_Pool { private: struct Task_Wrapper { ... }; atomic<bool> _done_; Blocking_Queue<Task_Wrapper> _queue_; // #1 unsigned _workersize_; thread* _workers_; void work() { Task_Wrapper task; while (!_done_.load(memory_order_acquire)) { _queue_.pop(task); // #3 task(); } } void stop() { while (!_queue_.empty()) // #2 std::this_thread::yield(); _done_.store(true, memory_order_release); for (unsigned i = 0; i < _workersize_; ++i) // #5 _queue_.push([] {}); for (unsigned i = 0; i < _workersize_; ++i) { if (_workers_[i].joinable()) _workers_[i].join(); // #4 } delete[] _workers_; } public: Thread_Pool() : _done_(false) { try { _workersize_ = thread::hardware_concurrency(); _workers_ = new thread[_workersize_]; for (unsigned i = 0; i < _workersize_; ++i) { _workers_[i] = thread(&Thread_Pool::work, this); } } catch (...) { stop(); throw; } } ~Thread_Pool() { stop(); } template<class Callable> future<typename std::result_of<Callable()>::type> submit(Callable c) { typedef typename std::result_of<Callable()>::type R; packaged_task<R()> task(c); future<R> r = task.get_future(); _queue_.push(std::move(task)); return r; } };

Task_Wrapper 具体化了线程安全的任务队列 Blocking_Queue<>（#1）。稍后对 Blocking_Queue<> 做说明。

终止线程池时，任务队列中的剩余任务被执行（#2）完后，任务队列处于被清空的状态。在 _done_ 还未被置为 true 之前，工作线程可能会因为 _queue_.pop(task) 而进入循环等待的阻塞状态（#3）。如果此时主线程先调用工作线程的 join() 函数（#4），将导致死锁（deadlock）状态。即，工作线程正在等待有任务入队，而主线程又要等待工作线程的结束。为了打破循环等待条件，在主线程调用工作线程的 join() 函数之前，向队列中放入 _workersize_ 个假任务（#5）。其目的，是确保在任务队列上等待的所有工作线程退出循环等待（#3）。

阻塞式的线程安全的任务队列 Blocking_Queue<> 如下，

template<class T> class Blocking_Queue { private: mutex mutable _m_; // #1 condition_variable _cv_; queue<T> _q_; public: void push(T&& element) { lock_guard<mutex> lk(_m_); _q_.push(std::move(element)); _cv_.notify_one(); // #3 } void pop(T& element) { unique_lock<mutex> lk(_m_); _cv_.wait(lk, [this]{ return !_q_.empty(); }); // #2 element = std::move(_q_.front()); _q_.pop(); } bool empty() const { lock_guard<mutex> lk(_m_); return _q_.empty(); } size_t size() const { lock_guard<mutex> lk(_m_); return _q_.size(); } };

队列采用了 std::mutex 和 std::condition_variable 控制工作线程对任务队列的并发访问（#1）。如果没有可出队的任务，当前线程就会在 _cv_ 上循环等待（#2）；任务入队后，由 _cv_ 通知在其上等待的线程（#3）。

逻辑

以下类图展现了此线程池的代码主要逻辑结构。

class