线程同步消息队列与线程池

发表于 分类于 ， 阅读次数：

线程同步消息队列与线程池

时间：2026/04/09

关键词：任务队列、worker thread、future、停止协议、背压、线程池
核心目标：理解线程池为什么几乎总是“队列 + 工作线程 + 生命周期管理”的组合。

1. 为什么线程池比“每个任务一个线程”更常见

直接为每个任务创建线程的问题在于：

  • 创建销毁开销高
  • 线程数不可控
  • 容易把系统调度器压爆

线程池的思路是：

  • 预先创建固定数量 worker
  • 任务进入共享队列
  • worker 从队列取任务执行

2. 线程池最小结构

一个线程池通常包含：

  • 任务队列
  • 多个工作线程
  • 停止标志
  • 提交接口

示意：

1
producer -> task queue -> workers

3. 推荐的任务表示

最常见的是：

1
std::function<void()>

这样线程池不关心任务具体类型，只负责执行。

如果要返回值，可以把真实任务包装进：

  • std::packaged_task
  • std::promise
  • std::future

4. 一个最小线程池骨架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

class ThreadPool {
public:
    explicit ThreadPool(std::size_t n) {
        for (std::size_t i = 0; i < n; ++i) {
            workers_.emplace_back([this] { worker_loop(); });
        }
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            stop_ = true;
        }
        cv_.notify_all();
        for (auto& t : workers_) {
            if (t.joinable()) t.join();
        }
    }

    void submit(std::function<void()> task) {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            tasks_.push(std::move(task));
        }
        cv_.notify_one();
    }

private:
    void worker_loop() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lk(mutex_);
                cv_.wait(lk, [&] { return stop_ || !tasks_.empty(); });
                if (stop_ && tasks_.empty()) return;
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            task();
        }
    }

    bool stop_ = false;
    std::mutex mutex_;
    std::condition_variable cv_;
    std::queue<std::function<void()>> tasks_;
    std::vector<std::thread> workers_;
};

5. 为什么停止协议很重要

如果没有明确的停止逻辑，线程池很容易在析构时：

  • worker 永远等在 wait
  • 主线程 join 不回来

正确退出条件通常是：

  • stop_ == true
  • 并且队列已空

6. 返回值怎么做

常见写法是：

  • 把用户任务包装成 packaged_task
  • 返回对应 future

这样提交方既能异步执行，也能之后 get() 结果。

线程池的接口常见长这样：

1
2
template <class F, class... Args>
auto enqueue(F&& f, Args&&... args) -> std::future<...>;

这也是完美转发的高频实战场景。

7. 有界任务队列与背压

如果任务生产速度远大于消费速度，线程池也可能把内存吃爆。
所以工程上经常要考虑：

  • 队列容量上限
  • 超限后阻塞
  • 超限后丢弃
  • 超限后降级

这其实就是背压策略。

8. 线程池不是越多线程越好

线程数通常取决于：

  • CPU 核心数
  • 任务是否 CPU 密集
  • 任务是否经常阻塞 I/O

经验上：

  • CPU 密集型：线程数通常接近核心数
  • I/O 密集型：线程数可适当更大

9. 消息队列 vs 线程池

这两个概念经常一起出现，但不完全一样。

  • 消息队列：强调数据传递与同步
  • 线程池：强调任务执行与线程复用

线程池内部几乎总会用到任务队列，但消息队列本身不一定等于线程池。

10. 常见坑

10.1 任务里抛异常没人管

如果没有 future 或显式捕获，异常可能直接导致线程终止。

10.2 析构时仍允许提交任务

这会让生命周期变得混乱。

10.3 持锁执行任务

这是严重错误。
正确做法是：

  • 取出任务后释放锁
  • 再执行任务

10.4 线程池里再无限提交内部任务

这可能制造级联膨胀和死锁风险。

11. 一页总结

线程池最关键的不是模板技巧，而是三个工程点：

  1. 任务队列
  2. worker 生命周期
  3. 明确的停止与背压策略

如果只记一句：

线程池本质上是“用受控线程数去消费一个受控任务流”。