队列空了该怎么等？写一个能正确退出的 C++ 阻塞队列

发表于 分类于 ， 阅读次数：
现代 C++ 实践 / 04更新于 2026-07-13预计阅读 8 分钟

时间：2026/04/09

生产者—消费者模型看起来很简单：一个线程把任务放进队列，另一个线程取出任务。最容易写出的消费者却往往是这样的：

1
2
3
4
5
6
7
while (running) {
    std::lock_guard<std::mutex> lock(mutex);
    if (!queue.empty()) {
        consume(queue.front());
        queue.pop();
    }
}

它至少有三个问题：队列为空时会持续占用 CPU；consume 在锁内执行会阻塞生产者；程序准备退出时，running 与队列中的剩余任务也没有清晰关系。

改用条件变量后，忙等虽然消失了，但新的问题随之出现：虚假唤醒怎么办？队列满时生产者是否可以无限堆积？消费者正在等待时怎样通知它“以后不会再有任务”？本文将围绕这些问题，实现一个有容量上限、支持背压且能够排空后关闭的阻塞队列。

1. 生产者和消费者真正需要协调什么？

队列只是存放数据的容器。线程之间真正共享的是一组状态：

1
2
3
4
5
6
生产者 --push--> [ 有界队列 ] --pop--> 消费者
                    │      │
                 是否满  是否为空
                    │      │
                 等待空间 等待数据
                    └──关闭状态──┘

一个完整的队列至少要定义三件事：

  1. 互斥访问：队列和关闭标志不能被多个线程无同步地读写；
  2. 条件等待：队列为空时消费者休眠，队列满时生产者等待；
  3. 关闭协议：停止接收新任务，并让消费者处理完已有任务后退出。

日志系统、线程池、网络消息分发都在使用这套结构。它们的差别主要在任务类型、过载策略和关闭语义，而不是“有没有一个 std::queue”。

2. 为什么条件变量不能只等一次通知？

下面的写法存在隐患：

1
2
condition.wait(lock);
auto value = queue.front();

线程从 wait 返回，不代表队列此刻一定非空，原因有两个：

  • 条件变量允许虚假唤醒（spurious wakeup）；
  • 多个消费者被唤醒后，另一个线程可能先取走唯一的元素。

条件变量传递的是“状态可能变化了”的通知，不保存“队列一定有数据”的事实。事实必须在同一把互斥锁保护下重新检查。

1
2
3
condition.wait(lock, [&] {
    return !queue.empty();
});

谓词版本等价于反复执行“条件不满足就继续等待”。锁在检查谓词时持有，在睡眠期间自动释放，醒来后重新取得。这使检查状态与进入等待之间不会丢失关键变化。

关闭状态也必须进入谓词，否则空队列上的消费者永远没有退出条件：

1
2
3
not_empty.wait(lock, [&] {
    return closed || !queue.empty();
});

3. 为什么无界队列会掩盖过载？

如果生产者平均每秒提交 10 万个任务，消费者只能处理 8 万个，积压就会以每秒 2 万个的速度增长。无界队列没有消除速度差，只是把它转换成不断增加的内存占用和越来越高的延迟。

有界队列（bounded queue）为积压设置上限。队列满时，生产者可以：

  • 阻塞等待，形成背压；
  • 立即返回失败，由调用者降级或重试；
  • 按业务规则丢弃新任务或旧任务。

本文选择第一种：push 在队列满时等待。它适合不能静默丢任务、且允许生产者减速的场景；它不适合事件循环等绝不能阻塞的线程。

4. 怎样定义一个不会卡住的关闭协议？

一个实用且容易理解的语义是“关闭输入，排空输出”：

  • close() 之后不再接受 push；
  • 已经入队的任务仍可被消费；
  • 队列关闭且为空时，pop() 返回 std::nullopt；
  • 所有正在等待的生产者和消费者都会被唤醒。

状态转换如下：

1
2
3
4
5
6
7
open + 有空间：push 成功
open + 已满：  push 等待
closed：       push 失败

有元素：       pop 返回元素（即使已经 closed）
open + 空：    pop 等待
closed + 空：  pop 返回 nullopt，消费者退出

这比塞入一个特殊的 "EXIT" 值更通用。哨兵会占用业务值空间，而且有多个消费者时还要决定放入多少个哨兵；关闭状态则是队列协议的一部分。

5. 一个最小可运行的有界阻塞队列

下面的 C++17 示例包含完整的入队、出队、背压和关闭逻辑。生产者提交 1 到 5，消费者计算总和；主线程先等待生产者结束，再关闭队列，保证消费者能排空已有元素。

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
#include <condition_variable>
#include <cstddef>
#include <iostream>
#include <mutex>
#include <optional>
#include <queue>
#include <stdexcept>
#include <thread>
#include <utility>

template <class T>
class BlockingQueue {
public:
    explicit BlockingQueue(std::size_t capacity) : capacity_(capacity) {
        if (capacity_ == 0) {
            throw std::invalid_argument("capacity must be greater than zero");
        }
    }

    BlockingQueue(const BlockingQueue&) = delete;
    BlockingQueue& operator=(const BlockingQueue&) = delete;

    bool push(T value) {
        std::unique_lock<std::mutex> lock(mutex_);
        not_full_.wait(lock, [this] {
            return closed_ || queue_.size() < capacity_;
        });

        if (closed_) {
            return false;
        }

        queue_.push(std::move(value));
        lock.unlock();
        not_empty_.notify_one();
        return true;
    }

    std::optional<T> pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        not_empty_.wait(lock, [this] {
            return closed_ || !queue_.empty();
        });

        if (queue_.empty()) {
            return std::nullopt;
        }

        T value = std::move(queue_.front());
        queue_.pop();
        lock.unlock();
        not_full_.notify_one();
        return value;
    }

    void close() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            closed_ = true;
        }
        not_empty_.notify_all();
        not_full_.notify_all();
    }

private:
    const std::size_t capacity_;
    std::queue<T> queue_;
    bool closed_ = false;
    std::mutex mutex_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
};

int main() {
    BlockingQueue<int> queue(2);
    int sum = 0;

    std::thread consumer([&] {
        while (auto value = queue.pop()) {
            sum += *value;
        }
    });

    std::thread producer([&] {
        for (int value = 1; value <= 5; ++value) {
            if (!queue.push(value)) {
                return;
            }
        }
    });

    producer.join();
    queue.close();
    consumer.join();

    std::cout << "sum = " << sum << '\n';
}

在 macOS、Linux 或支持 C++17 线程库的平台上编译：

1
2
3
clang++ -std=c++17 -O2 -Wall -Wextra -pedantic -pthread \
  blocking_queue.cpp -o blocking_queue
./blocking_queue

预期输出：

1
sum = 15

线程调度顺序并不确定，但最终求和结果确定。sum 只由消费者线程写入，并且主线程在 consumer.join() 后才读取，所以这里不存在数据竞争。

6. 这段代码为什么能够正确唤醒和退出？

push 等待的不是某次通知，而是“队列已经关闭，或者出现可写空间”。如果关闭，它立即返回 false；否则入队，并通知一个等待数据的消费者。

pop 的谓词同样包含两种可能：队列非空，或者队列已关闭。醒来后，它优先取走已有元素，因此关闭不会丢弃积压任务；只有队列确实为空时才返回 nullopt。

close 使用 notify_all，因为此时改变的是影响所有等待者的全局状态：

  • 等在 not_empty_ 上的消费者都需要检查是否应该退出；
  • 等在 not_full_ 上的生产者都需要发现队列不再接受任务。

普通入队和出队使用 notify_one，因为一次状态变化通常只新增一个可消费元素或一个可写位置。通知放在解锁之后，可以减少被唤醒线程立刻阻塞在同一把锁上的机会；正确性仍来自受锁保护的状态与谓词，而不是通知先后本身。

7. 为什么消费任务必须放在队列锁之外？

pop() 在返回前已经从队列取走元素并释放锁，消费者随后才处理任务：

1
2
3
while (auto task = queue.pop()) {
    process(*task); // 此时没有持有 queue 内部的锁
}

如果把 process 放在队列锁内，一次慢 I/O 或长计算就会阻止其他生产者入队，也会阻止其他消费者取任务。队列的锁应该只保护队列状态，不应该扩大到业务处理过程。

同理，不要在持有队列互斥锁时调用可能回调用户代码的操作。锁的临界区越小且职责越单一，死锁和长时间争用越容易控制。

8. 还需要哪些接口和策略？

真实项目可能需要 try_push、try_pop 或带超时的操作。它们解决的是不同调用约束：

接口 队列暂时不可操作时 适合场景
push / pop 阻塞等待 后台工作线程
try_push / try_pop 立即返回失败 事件循环、轮询逻辑
push_for / pop_for 最多等待一段时间 需要超时预算的请求链路

添加这些接口时，仍应复用同一组共享状态和关闭语义，避免不同函数对“关闭后是否排空”给出相互矛盾的答案。

对于 T 的异常行为也要明确：示例在锁内移动队首元素。如果 T 的移动构造会抛异常，pop 的异常保证取决于 T 和底层容器。任务队列通常选择可无异常移动的任务包装，或针对业务类型设计更明确的错误策略。

9. 动态扩容真的是好主意吗？

有界队列的容量是背压策略的一部分。队列一满就自动扩容，会逐渐把它变回无界队列，掩盖消费者跟不上生产者的事实。

如果业务确实需要在运行时增大容量，必须在保护队列状态的同一把锁内修改容量，并使用 not_full_.notify_all() 唤醒可能同时获得新位置的生产者。还要设置最大值并处理整数溢出。

缩容更谨慎：新容量不能小于当前元素数量，否则队列会立即进入“元素数已经超过容量”的状态。多数系统更适合在创建队列时根据延迟预算和可用内存确定固定容量，并通过监控排队长度调整配置，而不是让队列自行增长。

10. 工程中最容易踩哪些坑？

误区一：用 if 检查一次条件就继续

虚假唤醒和多消费者竞争都会让条件再次失效。始终使用谓词版 wait，或显式 while 循环重新检查受锁保护的状态。

误区二：析构队列会自动唤醒等待线程

如果仍有线程在成员条件变量上等待，直接析构队列属于生命周期错误。拥有队列和线程的上层对象必须先 close()，再等待所有线程 join()，最后才销毁队列。

误区三：用原子 running 就足够退出

消费者可能睡在条件变量里，看不到原子值变化。改变退出状态后仍需通知等待线程，而且退出状态必须与队列排空规则一起判断。

误区四：任何线程都可以阻塞在 push

如果 push 运行在负责继续消费或处理网络事件的唯一线程上，满队列可能形成系统级死锁或停顿。这种线程应使用非阻塞提交、超时或明确的丢弃策略。

误区五：通知可以替代共享状态

条件变量不是消息队列，通知可能在没有等待者时发生。线程能否继续必须由互斥锁保护的谓词决定，通知只负责促使它重新检查。

11. 什么时候适合使用阻塞队列？

阻塞队列适合线程之间传递独立任务，并且允许提交者或消费者在没有进展时休眠的场景，例如后台日志写入、固定工作线程池和离线流水线。

它不一定适合：

  • 单线程事件循环中的热路径；
  • 极低延迟且无法接受互斥竞争的系统；
  • 任务之间存在复杂依赖、取消和优先级调度；
  • 只需传递一个最新状态而非保存每条消息的场景。

这些场景可能需要异步提交、专门调度器、原子状态或其他数据结构。不要因为“跨线程通信”就默认套用同一种队列。

12. 总结

开头的忙等消费者真正缺少的不是一次 sleep，而是一套完整的状态协议：何时等待、由什么状态唤醒、积压怎样限制、关闭后剩余任务如何处理。

  1. 互斥锁保护队列、容量和关闭标志这一组共享状态；
  2. 条件变量必须与谓词检查结合，以应对虚假唤醒和线程竞争；
  3. 容量上限提供背压，不能把过载简单转化成无限内存增长；
  4. close() 应唤醒所有等待者，并明确是排空还是丢弃已有任务；
  5. 上层必须遵循“停止提交 → 关闭队列 → join 线程 → 销毁资源”的顺序。

实现生产者—消费者系统时，先写下关闭状态机，再写 push 和 pop。能正常传递任务只是起点，能在空队列、满队列和程序退出时都可靠前进，才算真正完成。