[C++知识库]C++ 简单线程池

今天看了github上点击量最大的一个C++2.0实现的线程池：链接，感觉看起来很简单明了
于是看后自己默写了一个线程池类，并对里面步骤进行注释，也是提升我自己的一部分
tips：原代码中的std::result_of在C++17的类型萃取库中已经移除，因此我改用指代更为清晰的decltype进行修改。

具体代码如下，代码风格采用谷歌式：
这大概是CSDN上面最优雅的C++线程池代码了吧

#pragma once

#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

class ThreadPool final {
 public:
  ThreadPool(uint8_t _num = 0);
  ~ThreadPool();
    
  ThreadPool(ThreadPool const&) = delete;
  ThreadPool(ThreadPool&&) = delete;
  ThreadPool& operator=(ThreadPool const&) = delete;
  ThreadPool& operator=(ThreadPool&&) = delete;

  // 函数模板声明在class体中，调用处实例化
  // 采用返回值后置，因为前置时形参还没声明，无法推导类型
  template <class F, class... Args>
  auto enqueue(F&& _func, Args&&... _args)
      -> std::future<decltype(_func(_args...))> {
    if (this->stop_.load()) {
      throw std::runtime_error("enqueue on stopped ThreadPool");
    }
    
    // 假如入队的函数为：int(int, char*, std::string)
    // decltype(_func(_args...))表示函数执行后返回值的类型
    // 那么 using rtype = int
    using rtype = decltype(_func(_args...));

    // 通过std::bind创造返回值相同的无参函数
    // 使用std::forward对入参进行完美转发
    // 从而得到对函数的包装对象：std::packaged_task<int()>
    auto task = std::make_shared<std::packaged_task<rtype()>>(
        std::bind(std::forward<F>(_func), std::forward<Args>(_args)...));

    // std::future<int>
    std::future<rtype> res = task->get_future();

    {
      std::lock_guard<std::mutex> lock(mtx_);
      
      // 由于tasks_中调用对象类型为void()
      // 构造一个返回值为void且无参数的匿名函数，在函数体中执行类型为int()的函数
      this->tasks_.emplace([task]() { (*task)(); });
    }
      
    // 不敢保证notify_one只会通知一个消费者而不会通知到自己，暂时我稍作改动，采取惊群
    this->cv_.notify_all();
    return res;
  }

 private:
  std::atomic<bool> stop_;
  std::vector<std::thread> threads_;
  std::queue<std::function<void()>> tasks_;
  std::mutex mtx_;
  std::condition_variable cv_;
};

#include "thread_pool.hpp"

ThreadPool::ThreadPool(uint8_t _num) : stop_{false} {
  for (auto i{0}; i != _num; ++i) {
    this->threads_.emplace_back([this] {
      while (true) {
        auto task = std::function<void()>();

        {
          std::unique_lock<std::mutex> lock(this->mtx_);
		 // 停止时不阻塞，运行时任务队列非空时不阻塞
          this->cv_.wait(lock, [this] {
            if (this->stop_.load() or !this->tasks_.empty()) {
              return true;
            }
            return false;
          });
	
          // 保证受到停止信号后，消费完所有任务
          if (this->stop_.load() and this->tasks_.empty()) {
            break;
          }

          task = std::move(this->tasks_.front());
          this->tasks_.pop();
        }

        task();
      }
    });
  }
}

ThreadPool::~ThreadPool() {
  this->stop_.store(true);
  this->cv_.notify_all();

  for (auto&& thread : this->threads_) {
    if (thread.joinable()) {
      thread.join();
    }
  }
}

运用：

#include "thread_pool.hpp"
 
using namespace std::literals;

int main() {
  // 根据硬件支持的最大并发数创建线程池
  auto pool = new ThreadPool(std::thread::hardware_concurrency());
  auto func = [](int a, int b) -> int { return std::pow(a, b); };
  auto futures = std::vector<std::future<int>>();

  for (auto i{0}; i ！= 10; ++i) {
    futures.push_back(pool->enqueue(func, i, 2));
  }

  // do something
  std::this_thread::sleep_for(10ms);

  for (auto&& future : futures) {
    std::cout << future.get() << std::endl;
  }
  delete pool;

  return 0;
}