C++八股 —— 手撕线程池

文章目录

• • 一、背景
  • 二、线程池实现
  • • 1. 任务队列和工作线程
    • 2. 构造和析构函数
    • 3. 添加任务函数
    • 4. 完整代码
    • 三、阻塞队列实现
    • • 1. 基础队列
      • 2. 升级版队列
      • 四、测试代码
      • 五、相关问题
      • 六、其他实现方式

        来自：华为C++一面：手撕线程池_哔哩哔哩_bilibili

        华为海思：

        手撕线程池

        相关概念参考：

        • C++八股——函数对象、Lambda、bind、function_c++八股文-CSDN博客
        • C++ 11 lock_guard 和 unique_lock_lockguard-CSDN博客
        • explicit关键字
        • C++ —— 可变参数_c++ 可变参数-CSDN博客
        • 阻塞队列（超详细易懂）-CSDN博客

          一、背景

          1. 什么是线程池

             维持管理一定数量线程的池式结构。

             核心思想：线程复用。 避免频繁地创建和销毁线程带来的开销。

          2. 为什么需要线程池

             • 创建/销毁线程的开销大，线程池可以有效降低资源消耗、提高响应速度
             • 提高线程的可管理性
             • 防止因任务过多导致无限制创建线程而耗尽系统资源的问题。

             • 线程池的工作流程

               核心为生产者-消费者模型

               

               线程池需要维护工作线程（消费者线程）和一个任务队列，生产者线程创建任务放入线程池的任务队列，消费者线程从任务队列中取出任务执行。

          二、线程池实现

          一个线程池包含：

          • 任务队列：存放生产者线程创建的任务
          • 工作线程：取出任务队列中任务执行
          • 构造函数
          • 析构函数
          • 添加任务函数
          • 工作线程函数

            1. 任务队列和工作线程

            任务队列使用一个手动实现的阻塞队列来实现；

            工作线程使用一个线程vector来实现。

            BlockingQueuePro task_queue_; // 任务队列
            std::vector workers_; // 工作线程列表
            

            ---

            工作线程函数是一个不断循环的函数，从任务队列中取出任务并执行

            // 工作线程函数
            void Worker() {
                while (true) {
                    std::function task;
                    if (!task_queue_.Pop(task))
                        break;
                    task(); // 执行任务
                }
            }
            

            2. 构造和析构函数

            构造函数传入一个整数作为线程池最大线程数，然后创建该数量的线程

            // 构造函数
            explicit ThreadPool(int num_threads) {
                for (size_t i = 0; i  
            析构函数将阻塞队列设置为非阻塞模型，并阻塞当前线程等待所有工作线程执行完毕
             
            // 析构函数
            ~ThreadPool() {
                task_queue_.Cancel();
                for (auto &worker : workers_) {
                    if (worker.joinable()) {
                        worker.join();
                    }
                }
            }
             
            3. 添加任务函数
             
            Post函数传入一个可调用对象和参数，将可调用对象和参数绑定之后加入到工作队列中。
             
            // 添加任务
            template 
            void Post(F &&f, Args &&...args) {
                auto task = std::bind(std::forward(f), std::forward(args)...);
                task_queue_.Push(task);
            }
             
            4. 完整代码
             
            class ThreadPool {
            public:
                // 构造函数
                explicit ThreadPool(int num_threads) {
                    for (size_t i = 0; i  
            三、阻塞队列实现
             
            阻塞队列是一种特殊的队列，同样遵循“先进先出”的原则，支持入队操作和出队操作。在此基础上，阻塞队列会在队列已满或队列为空时陷入阻塞，使其成为一个线程安全的数据结构，它具有如下特性：
             
            

            • 当队列已满时，继续入队列就会阻塞，直到有其他线程从队列中取走元素。
            • 当队列为空时，继续出队列也会阻塞，直到有其他线程向队列中插入元素。

              （引用参考：阻塞队列（超详细易懂）-CSDN博客）

              1. 基础队列

              生产者和消费者共用一个队列和互斥锁

              • 当队列为空时，使工作线程进入休眠。
              • 当队列被设置为非阻塞时，队列任务为空会使工作线程结束

                源码：

                template 
                class BlockingQueue {
                public:
                    BlockingQueue(bool nonblock = false) : nonblock_(nonblock) {}
                    // 添加任务
                    void Push(const T &task) {
                        std::lock_guard lock(mutex_);
                        queue_.push(task);
                        not_empty_.notify_one(); // 通知一个等待的线程
                    }
                    // 获取任务
                    bool Pop(T &task) {
                        std::unique_lock lock(mutex_);
                        not_empty_.wait(lock, [this] { return !queue_.empty() || nonblock_; });
                        if (queue_.empty()) 
                            return false; 
                        task = queue_.front();
                        queue_.pop();
                        return true;
                    }
                    // 解除阻塞当前队列的线程
                    void Cancel() {
                        std::lock_guard lock(mutex_);
                        nonblock_ = true; // 设置为非阻塞状态
                        not_empty_.notify_all(); // 通知所有等待的线程
                    }
                private:
                    bool nonblock_; // 是否为非阻塞模式
                    std::mutex mutex_; // 互斥锁
                    std::condition_variable not_empty_; // 条件变量，队列为空时线程休眠
                    std::queue queue_; // 任务队列
                };
                

                2. 升级版队列

                生产者和消费者有各自的任务队列和互斥锁

                • 当消费者队列为空时，会尝试与生产者队列交换

                  • 若交换中生产者队列为空，使工作线程进入休眠；

                  • 若队列被设置为非阻塞，生产者队列为空，交换后消费者队列仍为空，此时会结束工作线程。

                    源码：

                    // 升级版队列，多生产者和多消费者
                    template 
                    class BlockingQueuePro {
                    public:
                        BlockingQueuePro(bool nonblock = false) : nonblock_(nonblock) {}
                        // 添加任务
                        void Push(const T &task) {
                            std::lock_guard lock(producer_mutex_);
                            producer_queue_.push(task);
                            not_empty_.notify_one(); // 通知一个等待的线程
                        }
                        // 获取任务
                        bool Pop(T &task) {
                            std::unique_lock lock(consumer_mutex_);
                            // 如果消费者队列为空，尝试交换生产者队列
                            if (consumer_queue_.empty() && SwapQueue_() == 0) {
                                return false; // 如果交换后仍然为空，则返回false
                            }
                            task = consumer_queue_.front();
                            consumer_queue_.pop();
                            return true;
                        }
                        // 解除阻塞当前队列的线程
                        void Cancel() {
                            std::lock_guard lock(producer_mutex_);
                            nonblock_ = true; // 设置为非阻塞状态
                            not_empty_.notify_all(); // 通知所有等待的线程
                        }
                    private:
                        // 交换生产者队列到消费者队列
                        size_t SwapQueue_() {
                            std::unique_lock lock(producer_mutex_);
                            not_empty_.wait(lock, [this] { return !producer_queue_.empty() || nonblock_; });
                            std::swap(producer_queue_, consumer_queue_); // 交换队列
                            return consumer_queue_.size(); // 返回新的消费者队列大小
                        }
                        bool nonblock_; // 是否为非阻塞模式
                        std::mutex producer_mutex_; // 生产者互斥锁
                        std::mutex consumer_mutex_; // 消费者互斥锁
                        std::condition_variable not_empty_; // 条件变量，队列为空时线程休眠
                        std::queue producer_queue_; // 生产者任务队列
                        std::queue consumer_queue_; // 消费者任务队列
                    };
                    

                    四、测试代码

                    • 任务函数Task()：线程池中工作线程需要执行的任务
                    • 生产者函数Producer()：将num_tasks个任务添加到线程池中
                    • 生产者线程producers：包含多个生产者，同时并行生成任务到线程池中
                    • 等待生产者线程完成任务生成
                    • 等待线程池执行完所有生成的任务

                      #include 
                      #include 
                      #include 
                      #include 
                      #include 
                      #include "threadpool.h"
                      // 全局计数器，统计任务完成的数量
                      std::atomic task_counter(0);
                      // 任务函数
                      void Task(int id) {
                          std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟任务处理时间
                          std::cout 
                          for (int i = 0; i