从零实现一个轻量级C++线程池

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <future>
#include <mutex>
#include <functional>
#include <chrono>
#include <type_traits>
#include <condition_variable>
class ThreadPool {
public:
    ThreadPool(size_t thread_num = 4):
        _thread_num(thread_num),
        _start(false),
        _stop(false)
    {}
    ~ThreadPool(){
        if (_start && !_stop) stop();
    }
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool& operator=(ThreadPool&&) = delete;
    void start() {
        std::unique_lock<std::mutex> lock(_mutex);
        if (_start) return;
        _workers.reserve(_thread_num);
        for (size_t i = 0; i < _thread_num; i++) {
            _workers.emplace_back(std::thread([this](){
                work_loop();
            }));
        }
        _start = true;
    }
    void stop() {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_start || _stop) return;
            // 在join回收线程之前，必须先将_stop置为true，
            // 否则工作线程可能会一直阻塞在条件变量上，导致无法正常退出，甚至会导致程序崩溃
            _stop = true;
        }
        _cond.notify_all();
        for (auto& worker : _workers) {
            if (worker.joinable()) worker.join();
        }
    }
    template<class F, class... Args>
    auto submit(F&& f, Args&&... args)->std::future<std::invoke_result_t<F, Args...>> {
        using return_type = std::invoke_result_t<F, Args...>;
        // 绑定函数参数，并交给任务包装器
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        // 获取关联的future
        std::future<return_type> res = task->get_future();
        // 加锁+入队列
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_stop || !_start) throw std::runtime_error("线程池未启动！");
            _tasks.emplace([task](){(*task)();});
        }
        _cond.notify_one();
        return res;
    }
private:
    void work_loop() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock, [this](){
                    return _stop || !_tasks.empty();
                });
                if (_stop && _tasks.empty()) return;
                task = std::move(_tasks.front());
                _tasks.pop();
            }
            task();
        }
    }
private:
    std::vector<std::thread> _workers;
    std::queue<std::function<void()>> _tasks;
    std::mutex _mutex;
    std::condition_variable _cond;
    size_t _thread_num;
    bool _start;
    bool _stop;
};
int add(int a, int b) {
    return a + b;
}
void print() {
    std::cout << "-------------------print-------------------" << std::endl;
    std::cout << "Hello World!" << std::endl;
}
int main() {
    ThreadPool pool(4);
    pool.start();
    std::cout << "==================ThreadPoolTest==================" << std::endl;
    pool.submit([](){
        std::cout << "-------------------lambda-------------------" << std::endl;
        std::cout << "this is a lambda!" << std::endl;
    });
    std::this_thread::sleep_for(std::chrono::seconds(3));  
    auto ret1 = pool.submit(add, 10, 20);
    std::cout << "-------------------add-------------------" << std::endl;
    std::cout << "10 + 20 = " << ret1.get() << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(3));  
    pool.submit(print);
    pool.stop();
    return 0;
}