Thread Pool 线程池

本文基于经典的99行代码，稍加修改使其支持C++20，并增加了wait与join功能。

#pragma once
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

namespace ThreadPool {
	class ThreadPool {
	public:
		ThreadPool(size_t);
		template
		auto enqueue(F&& f, Args&&... args)
			->std::future::type>;
		~ThreadPool();
		void wait();
		void join();
	private:
		std::vector workers; //线程池
		std::queue> tasks; //任务队列
		std::mutex queue_mutex; //队列锁
		std::condition_variable condition; //状态变量
		std::atomic stop; //原子变量
		std::vector status;
	};

	// the constructor just launches some amount of workers
	inline ThreadPool::ThreadPool(size_t threads) : stop(false), status(threads, 0)
	{
		for (size_t i = 0; i < threads; ++i)
		{
			workers.emplace_back( //增加工作线程
				[this, i]
				{
					for (;;) //线程死循环
					{
						std::function task; //函数对象

						{
							std::unique_lock 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();
							//std::cout << "Thread " << i << " is working" << std::endl;
							//std::cout << "The task size is " << this->tasks.size() << std::endl;
						}
						this->status[i] = true;
						task(); //执行任务
						this->status[i] = false;
					}
				}
				);
		}
	}

	// add new work item to the pool
	template 
	auto ThreadPool::enqueue(F&& f, Args&&... args)
		-> std::future::type> //->运算符表示返回类型放在后方，返回类型为std::future类型，且由模板std::invoke_result::type决定
	{
		using return_type = typename std::invoke_result::type; //重命名返回类型

		auto task = std::make_shared>(
			std::bind(std::forward(f), std::forward(args)...)
			);//构造一个函数对象智能指针，std::package_task可以包装一个异步返回的函数，std::forward可以进行完美转发。

		std::future res = task->get_future(); //res保存线程返回值
		{
			std::unique_lock 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;
	}

	// the destructor joins all threads
	inline ThreadPool::~ThreadPool()
	{
		{
			std::unique_lock lock(queue_mutex);
			stop = true;
		}
		condition.notify_all(); //唤醒所有线程
		for (auto& worker : workers)
		{
			if (worker.joinable()) {
				worker.join();
			}
		}
	}

	inline void ThreadPool::wait()
	{
		bool isRuning = true;
		while (isRuning) {
			isRuning = false;
			for (auto& status : this->status) {
				if (status) {
					isRuning = true;
					break;
				}
			}
			std::this_thread::sleep_for(std::chrono::milliseconds(100));
		}
	}

	inline void ThreadPool::join()
	{
		{
			std::unique_lock lock(queue_mutex);
			stop = true;
		}
		condition.notify_all();
		for (auto& worker : workers)
		{
			if (worker.joinable()) {
				worker.join();
			}
		}
	}
}

进行了一定的完善，支持在C++20中使用，增加了join与wait函数。
用法：

// 创建一个4个工作线程的线程池
ThreadPool pool(4);

// 保存返回结果，std::future
auto result = pool.enqueue([](int answer) { return answer; }, 42);

// 等待结果并输出
std::cout << result.get() << std::endl;

// 创建一个4个工作线程的线程池
ThreadPool pool(4);
// 创建一个future的vector批量保存结构
std::vector> results;

for(int i = 0; i < 10; i++){
    // 保存返回结果，std::future
    results.emplace_back(pool.enqueue([](int answer) { return answer*answer; }, i));
}

//输出结果
for(auto& res : results){
    std::cout << res.get() << std::endl;
}
results.clear();

GitHub: Thread Pool