实现一个线程池

2022-09-06

为什么要使用线程池

并发原理，其实就是cpu快速来回切换，在特定的时间执行特定的某一个任务。并发执行存在着线程间上下文切换的问题，会消耗一定的时间。如果不考虑阻塞，多线程并发执行其实比单线程执行更加耗费时间，线程过多也会造成CPU负荷过大，并且线程占用内存资源，创建销毁线程也都是需要开销的.

如果在对到来的事件都显式的创建线程来处理,频繁的创建和销毁线程带来巨大的开销.

在线程池中提前启动若干线程,线程处于睡眠态.当有处理任务时唤醒线程处理任务.这样就使得更多的程序运行时间来处理实际的任务,而不是频繁地创建和销毁线程.

单任务队列线程池

线程池主要由任务队列和线程池(这里的池指的是一组资源的集合)组成.任务添加到任务队列,线程池从任务队列中取出任务进行处理.

如何把任务队列和线程池组合在一起?

简易单任务队列线程池的成员只有一个线程组即我们的线程池，一个任务队列.为了保证任务队列的线程安全，我们还提供了一个锁.同时我们还提供了一个条件变量，利用锁和条件变量，我们可以实现线程通知机制.当一个任务进入到线程池中，就会通知一个线程去处理到来的任务。

如何把任务添加到任务队列?

留出enqueue接口.当需要添加任务时.调用此接口,将任务添加至任务队列.

如何从任务队列中取出任务交由线程处理?

在把任务添加至任务队列的同时可以通过条件变量唤醒在阻塞状态下的池中资源,池中资源自动从,任务队列头取出任务并执行.

如何用cpp11实现

从github上找到 https://github.com/progschj/ThreadPool cpp11单任务队列线程池的实现.这里我加上了注释,任然有很多cpp11的语法糖让人比较费解.

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


// * 简易多线程单任务队列线程池，使用线程安全队列，接口更人性化。
class ThreadPool {
public:
    explicit ThreadPool(size_t threads = std::thread::hardware_concurrency()) : stop(false) {
        // 根据threads数量创建多个线程
        for (size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this]() {
                for (;;) {// 工作线程就是一个死循环，不停查询任务队列并取出任务执行
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock,
                                             [this]() {
                                                 return this->stop || !this->tasks.empty();// 条件变量等待线程池不为空或者stop
                                             });
                        if (this->stop && this->tasks.empty())// 线程池为空且stop，证明线程池结束，退出线程
                            return;
                        task = std::move(this->tasks.front());// 取出任务
                        this->tasks.pop();
                    }

                    task();// 执行任务
                }
            });// lambda表达式构建
        }
    }

    template<typename F, typename... Args>
    auto enqueue(F &&f, Args &&...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)...));// 完美转发，构造任务仿函数的指针
        std::future<return_type> res = task->get_future();                  // 获得函数执行的future返回
        {
            std::unique_lock<std::mutex> lock(queue_mutex);

            if (stop) {
                throw std::runtime_error("enqueue on stopped Thread pool");
            }

            tasks.emplace([task = std::move(task)]() { (*task)(); });// 塞入任务队列
        }                                                            // 入队列后即可解锁
        condition.notify_one();                                      // 仅唤醒一个线程，避免无意义的竞争
        return res;
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();// 唤醒所有线程，清理任务
        for (std::thread &worker: workers)
            worker.join();// 阻塞，等待所有线程执行结束
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};