欢迎光临
我们一直在努力
        找回密码
当前位置：171主机测评 > 服务器技术 > 正文

c++线程池实现

2026-03-13 分类：服务器技术 阅读(86) 评论(0)

目录

1.什么是线程池？

2.线程池的组成

3.线程池解决的问题

4.线程池如何工作（简述）

5.线程数如何选择

6.为什么维持固定数量线程？

7.手写线程池（设计思路）

8.单队列线程池实现

8.1ThreadPool.h

8.2ThreadPool.cc

8.3BlockingQueue.h

9.双队列的线程池实现

9.1ThreadPool.h

9.2ThreadPool.cc

9.3BlockingQueuepro.h

10.测试代码

11.总结

1.什么是线程池？

线程池是一种维持并管理固定数量线程的资源池。通过复用线程来执行多个任务，避免频繁创建/销毁线程带来的开销，并提高并发任务处理效率。

2.线程池的组成

  • 生产者（Producer）：提交耗时或异步任务的线程或模块。

  • 任务队列（Task Queue）：用于存放任务上下文与回调函数（通常封装为 std::function<void()> 或类似类型）。

  • 消费者（Worker / 线程池中的线程）：从任务队列取出任务并执行，负责线程调度与生命周期管理。

3.线程池解决的问题

  • 异步执行耗时任务，避免阻塞主线程或核心工作线程。

  • 充分利用多核 CPU，通过并行执行提升吞吐量。

  • 限制并发线程数量，防止系统资源耗尽。

  • 降低线程频繁创建与销毁的开销，提高性能与稳定性。

4.线程池如何工作（简述）

生产者将任务提交到任务队列；线程池内部的固定数量线程不断从队列中取出任务并执行。通常配合条件变量或信号量实现任务等待与唤醒。

5.线程数如何选择

  • CPU 密集型任务（以计算为主）：线程数 ≈ CPU 核心数（或略多于核心数）。

  • I/O 密集型任务（以等待为主）：经验公式为 2 * CPU 核心数，或根据比例估算： (线程等待时间 + CPU 运算时间) * CPU 核心数 / CPU 运算时间。 选择目标：尽量保持 CPU 在高利用但不过载、避免大量上下文切换。

6.为什么维持固定数量线程？

  • 超过一定数量后，线程增多不会带来性能提升，反而增加上下文切换和系统开销。

  • 固定线程数量能稳定资源使用，降低创建/销毁开销，提升性能稳定性。

7.手写线程池（设计思路）

实现线程池通常包括三部分：接口设计、数据结构设计、具体编码实现。

  • 接口设计（对外 API）

    • 构造与析构：创建指定数量线程，优雅停止并加入线程。

    • submit / enqueue：提交任务（支持返回值和异常传播，如使用 std::future）。

    • shutdown / stop：停止接受新任务并优雅退出，或强制停止。

    • 配置接口：设置线程数量、队列容量、任务超时或拒绝策略（可选）。

  • 数据结构设计

    • 任务封装类型：例如 std::function<void()> 或模板化包装为可以返回值的任务。

    • 任务队列：线程安全队列（单队列或多队列策略），配合互斥锁（std::mutex）与条件变量（std::condition_variable）。

    • 状态标志：标识线程池运行/停止状态，便于在停止时唤醒所有等待线程。

  • 具体编码

    我们将分别实现单队列和双队列两种线程池，并对比其优缺点。

    • 8.单队列线程池实现

    设计思路：所有工作线程共享同一个任务队列，通过互斥锁和条件变量保证线程安全。生产者向队列中 push 任务，消费者（工作线程）从队列中 pop 任务执行。

    8.1ThreadPool.h

    #pragma once
    ​
    #include <functional>
    #include <vector>
    #include <thread>
    ​
    // 使用前置声明，避免包含blockingqueue.h头文件，避免依赖循环引用
    template <typename T>
    class BlockingQueue; // 使用前置声明只能使用指针或者引用
    ​
    class ThreadPool
    {
    public:
       explicit ThreadPool(int num_thred_num); // explicit 避免隐式转换
       ~ThreadPool();
    ​
       void Post(std::function<void()> task);
    ​
    private:
       void Worker();
       std::unique_ptr<BlockingQueue<std::function<void()>>> task_queue_; // unique_ptr防止拷贝构造
       std::vector<std::thread> workers_;
    };

    8.2ThreadPool.cc

    ​
    #include "threadpool.h"
    #include "blockingqueue.h"
    ​
    ThreadPool::ThreadPool(int num_thred_num)
    {
       task_queue_ = std::make_unique<BlockingQueue<std::function<void()>>>(); // 1. 创建任务队列
       for (size_t i = 0; i < num_thred_num; i++)
      {
           workers_.emplace_back([this]
                                  { Worker(); });
      }
    }
    ​
    ThreadPool::~ThreadPool()
    {
       task_queue_->Cancel(); // 1. 取消所有待处理任务
       for (auto &worker : workers_)// 2. 遍历所有工作线程
      {
           if (worker.joinable()) // 3. 检查线程是否可加入
          {
               worker.join();  // 4. 等待线程结束
          }
      }
    }
    ​
    void ThreadPool::Post(std::function<void()> task)
    {
       task_queue_->Push(task);
    }
    ​
    void ThreadPool::Worker()
    {
       while (true)
      {
           std::function<void()> task;
           if (!task_queue_->Pop(task))
          {
               break;
          }
           task();
      }
    }

    8.3BlockingQueue.h

    #pragma once
    ​
    #include <queue>
    #include <mutex>
    #include <condition_variable>
    #include <functional>
    ​
    template <typename T>
    class BlockingQueue // 单生产者
    {
    public:
       BlockingQueue(bool non_block = false) : non_block_(non_block) {}
    ​
       void Push(const T &value)
      {
           std::lock_guard<std::mutex> lock(mutex_);
           queue_.push(value);
           not_empty_.notify_one();
      }
    ​
       // 正常pop 弹出元素
       // 异常pop 取消弹出 返回false
       bool Pop(T &value)
      {
           std::unique_lock<std::mutex> lock(mutex_);
    ​
           // 1. mutex.unlock()
           // 2. queue_.empty() || !non_block_ 线程在wait阻塞
           // notfiy_one 或者 notify_all 唤醒
           // 3. 假设满足条件 mutex.lock()
           // 4. 不满足条件 回到2继续阻塞
           not_empty_.wait(lock, [this]
                          { return !queue_.empty() || non_block_; });
           if (queue_.empty())
               return false;
           value = queue_.front();
           queue_.pop();
           return true;
      }
    ​
       void Cancel()
      {
           std::lock_guard<std::mutex> lock(mutex_);
           non_block_ = true;
           not_empty_.notify_all();
      }
    ​
    private:
       bool non_block_; // 是否为非阻塞模式 默认为阻塞模式
       std::queue<T> queue_;
       std::mutex mutex_;
       std::condition_variable not_empty_; // 非空条件变量
    };
    ​

    9.双队列的线程池实现

    设计思路：将任务队列分为“生产者队列”和“消费者队列”，减少锁竞争。生产者只向 pro_queue_ 中 push，消费者从 cons_queue_ 中 pop。当 cons_queue_ 为空时，工作线程通过条件变量等待，并在唤醒后将 pro_queue_ 与 cons_queue_ 交换，从而批量获取任务。这种方式可以减少消费者对生产者队列的锁占用，提高并发性能。

    9.1ThreadPool.h

    #pragma once
    ​
    #include <functional>
    #include <vector>
    #include <thread>
    ​
    // 使用前置声明，避免包含blockingqueue.h头文件，避免依赖循环引用
    template <typename T>
    class BlockingQueue; // 使用前置声明只能使用指针或者引用
    ​
    class ThreadPool
    {
    public:
       explicit ThreadPool(int num_thred_num); // explicit 避免隐式转换
       ~ThreadPool();
    ​
       void Post(std::function<void()> task);
    ​
    private:
       void Worker();
       std::unique_ptr<BlockingQueue<std::function<void()>>> task_queue_; // unique_ptr防止拷贝构造
       std::vector<std::thread> workers_;
    };

    9.2ThreadPool.cc

    #include "threadpool.h"
    #include "blockingqueue.h"
    #include "BlockingQueuepro.h"
    ​
    ThreadPool::ThreadPool(int num_thred_num)
    {
       task_queue_ = std::make_unique<BlockingQueuepro<std::function<void()>>>(); // 1. 创建任务队列
       for (size_t i = 0; i < num_thred_num; i++)
      {
           workers_.emplace_back([this]
                                  { Worker(); });
      }
    }
    ​
    ThreadPool::~ThreadPool()
    {
       task_queue_->Cancel(); // 1. 取消所有待处理任务
       for (auto &worker : workers_)// 2. 遍历所有工作线程
      {
           if (worker.joinable()) // 3. 检查线程是否可加入
          {
               worker.join();  // 4. 等待线程结束
          }
      }
    }
    ​
    void ThreadPool::Post(std::function<void()> task)
    {
       task_queue_->Push(task);
    }
    ​
    void ThreadPool::Worker()
    {
       while (true)
      {
           std::function<void()> task;
           if (!task_queue_->Pop(task))
          {
               break;
          }
           task();
      }
    }
    ​

    9.3BlockingQueuepro.h

    #pragma once
    ​
    #include <queue>
    #include <mutex>
    #include <condition_variable>
    #include <functional>
    ​
    template <typename T>
    class BlockingQueuepro {
    public:
       BlockingQueuepro(bool non_block = false) : non_block_(non_block) {}
    ​
       void Push(const T &value) {
           std::lock_guard<std::mutex> lock(pro_mutex_);
           pro_queue_.push(value);
           not_empty_.notify_one(); // 通知一个消费者线程
      }
    ​
       bool Pop(T &value) {
           std::unique_lock<std::mutex> lock(cons_mutex_);
           if(cons_queue_.empty() && SwapQueue_() == 0) {
               return false;
          }
           value = cons_queue_.front();
           cons_queue_.pop();
           return true;
      }
    ​
       void Cancel() {
           std::lock_guard<std::mutex> lock(pro_mutex_);
           non_block_ = true;
           not_empty_.notify_all();
      }
    ​
    private:
    ​
       int SwapQueue_() {
           std::unique_lock<std::mutex> lock(pro_mutex_);
           not_empty_.wait(lock, [this] { return !pro_queue_.empty() || non_block_;});
           std::swap(pro_queue_, cons_queue_);
           return cons_queue_.size();
      }
       bool non_block_; // 是否为非阻塞模式 默认为阻塞模式
       std::queue<T> pro_queue_; // 生产者队列
       std::queue<T> cons_queue_; // 消费者队列
       std::mutex pro_mutex_; // 生产者互斥锁
       std::mutex cons_mutex_; // 消费者互斥锁
       std::condition_variable not_empty_; // 非空条件变量
    };

    10.测试代码

    下面是一个简单的测试程序，模拟多个生产者向线程池提交任务，并统计任务完成数量。

    #include <iostream>
    #include <thread>
    #include <vector>
    #include <atomic>
    #include <chrono>
    ​
    #include "threadpool.h"
    ​
    // 全局计数。用于统计任务完成的数量
    std::atomic<int> task_counter{0};
    ​
    // 任务函数
    void Task(int id) {
       std::this_thread::sleep_for(std::chrono::milliseconds(100));
       std::cout << "Task " << id << " completed by thread "
                 << std::this_thread::get_id() << std::endl;
       task_counter++; // 任务完成，计数加一
    }
    ​
    // 生产者线程函数
    void Producer(ThreadPool &pool, int produer_id, int num_tasks) {
       for (int i = 0; i < num_tasks; i++) {
           int task_id = produer_id * 1000 + i; // 生成唯一任务ID
           pool.Post([task_id]() { Task(task_id); }); // 提交任务到线程池
           std::cout << "Producer " << produer_id
                     << " posted Task " << task_id << std::endl;
      }
    }
    ​
    int main() {
       const int num_producers = 4; // 生产者线程数量
       const int num_tasks_per_producer = 10; // 每个生产者提交的任务数量
       const int num_threads_in_pool = 2; // 线程池中的线程数量
    ​
       ThreadPool pool(num_threads_in_pool); // 创建线程池
    ​
       std::vector<std::thread> producers; // 生产者线程容器
    ​
       // 启动生产者线程
       for (int i = 0; i < num_producers; i++) {
           // emplace_back直接在vector中构造对象
           producers.emplace_back(Producer, std::ref(pool), i, num_tasks_per_producer); // 启动生产者线程
           // 相当于：
           // producers.push_back(std::thread(producer, std::ref(pool), i, num_tasks_per_producer));
      }
    ​
       // 等待所有生产者线程完成
       for (auto &producer : producers) {
           producer.join();
      }
    ​
       // 等待所有任务完成
       while (task_counter < num_producers * num_tasks_per_producer) {
           std::this_thread::sleep_for(std::chrono::milliseconds(100));
      }
    ​
       std::cout << "All tasks completed. Total tasks: "
                 << task_counter << std::endl;
    ​
       return 0;
    }

    11.总结

    本文介绍了线程池的基本概念、设计思路，并提供了两种典型的 C++ 线程池实现：单队列和双队列。单队列实现简单直观，但锁竞争可能成为性能瓶颈；双队列通过分离生产者和消费者队列，减少了锁粒度，适用于高并发场景。读者可以根据实际需求选择合适的实现，并进一步扩展支持任务返回值、超时、拒绝策略等高级特性。

    https://github.com/0voice

    赞(0)
    未经允许不得转载：171主机测评 » c++线程池实现
    分享到： 更多 (0)
    标签：

    相关推荐

    评论 抢沙发

    取消

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址

    热门标签

    #人工智能(58546)#python(19890)#开发语言(16711)#java(15341)#运维(15252)#linux(14867)#大数据(14461)#服务器(12984)#算法(11687)#前端(10776)#网络(9295)#数据库(8815)#ai(8771)#c++(8106)#课程设计(6782)#学习(6706)#机器学习(6657)#区块链(5884)#OpenClaw(5385)#后端(5103)#spring boot(4978)#深度学习(4888)#c语言(4353)#毕业设计(4287)#数据结构(4168)#安全(4002)#docker(3909)#架构(3821)#javascript(3783)#大模型(3616)

    热门文章

    最新文章

    归档

    分类