C++11标准下的简单线程池实现

一个简单的线程池实现思路，支持设置线程数、最大任务队列大小，push_task为阻塞式任务提交，当任务队列已满的时候会阻塞任务提交，直到任务队列有空闲位置。很久以前写的代码了，质量不高，效率也不太行，不过可以在写测试代码做多线程支持的时候cv一下应急用：

#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>

class toThreadPool {
private:
	// 成员变量
	std::mutex mutex_;
	std::condition_variable cv_task_;
	std::condition_variable cv_push_;
	std::condition_variable cv_finished_;
	std::queue<std::function<void()>> tasks_;
	std::vector<std::thread> threads_;
	const size_t max_queue_size_;
	size_t running_threads_;
	bool stop_requested_;
	bool force_stop_;

	void worker_thread() {
		while (true) {
			std::function<void()> task;
			{
				std::unique_lock<std::mutex> lock(mutex_);
				cv_task_.wait(lock, [this] {
					return !tasks_.empty() || stop_requested_;
				});

				// 处理停止请求
				if (stop_requested_ && (force_stop_ || tasks_.empty())) {
					break;
				}

				// 获取任务
				task = std::move(tasks_.front());
				tasks_.pop();
				running_threads_++;
				cv_push_.notify_one();
			}

			// 执行任务
			task();

			{
				std::lock_guard<std::mutex> lock(mutex_);
				running_threads_--;
				if (tasks_.empty() && running_threads_ == 0) {
					cv_finished_.notify_all();
				}
			}
		}
	}

public:
	toThreadPool(size_t max_queue_size = 1)
		: max_queue_size_(max_queue_size ? max_queue_size : SIZE_MAX), running_threads_(0), stop_requested_(false), force_stop_(false) {
	}

	~toThreadPool() {
		if (!stop_requested_)
			force_stop();
	}

	void start(size_t thread_count = std::thread::hardware_concurrency()) {
		std::lock_guard<std::mutex> lock(mutex_);
		for (size_t i = 0; i < thread_count; ++i) {
			threads_.emplace_back(&toThreadPool::worker_thread, this);
		}
	}

	template <typename F, typename... Args>
	void push_task(F &&f, Args &&...args) {
		using ReturnType = typename std::invoke_result_t<F, Args...>;
		auto task = std::make_shared<std::packaged_task<ReturnType()>>(
			std::bind(std::forward<F>(f), std::forward<Args>(args)...));

		std::unique_lock<std::mutex> lock(mutex_);
		// 等待队列有空位
		cv_push_.wait(lock, [this] {
			return tasks_.size() < max_queue_size_ || stop_requested_;
		});

		if (stop_requested_) {
			throw std::runtime_error("push_task on stopped ThreadPool");
		}

		// 添加任务到队列
		tasks_.emplace([task] {
			(*task)();
		});
		cv_task_.notify_one();
	}

	void wait() {
		std::unique_lock<std::mutex> lock(mutex_);
		cv_finished_.wait(lock, [this] {
			return tasks_.empty() && (running_threads_ == 0);
		});
	}

	void stop() {
		{
			std::lock_guard<std::mutex> lock(mutex_);
			stop_requested_ = true;
		}
		cv_task_.notify_all();
		cv_push_.notify_all();

		for (auto &t : threads_) {
			if (t.joinable())
				t.join();
		}
	}

	void force_stop() {
		{
			std::lock_guard<std::mutex> lock(mutex_);
			force_stop_ = true;
			stop_requested_ = true;
			// 清空任务队列
			while (!tasks_.empty())
				tasks_.pop();
		}
		cv_task_.notify_all();
		cv_push_.notify_all();

		// Detach所有线程
		for (auto &t : threads_) {
			if (t.joinable())
				t.detach();
		}
		threads_.clear();
	}
};