OliverChiu
管理
浏览知识库目录

C++

C++ 条件变量详解

发布于 2026年7月22日

C++ 条件变量详解

条件变量用于等待某个共享状态发生变化。它允许线程在条件不满足时进入休眠，而不是持续轮询占用 CPU。

C++ 提供两种条件变量：

#include <condition_variable>

std::condition_variable
std::condition_variable_any

最常用的是：

std::condition_variable

1. 条件变量解决什么问题

1.1 忙等待的问题

假设消费者必须等待生产者准备数据。

最简单的方式是循环检查：

std::atomic<bool> ready{false};

void consumer()
{
    while (!ready.load()) {
    }

    use_data();
}

这种方式称为忙等待。等待期间，线程会持续占用 CPU。

加入休眠虽然能降低 CPU 占用：

while (!ready.load()) {
    std::this_thread::sleep_for(
        std::chrono::milliseconds(10)
    );
}

但又会产生新的问题：

  • 休眠时间太短，线程仍会频繁唤醒；
  • 休眠时间太长，会增加响应延迟；
  • 很难确定合适的等待时间；
  • 本质上仍然是在轮询。

1.2 条件变量的作用

使用条件变量后，消费者可以在条件不满足时休眠：

std::mutex mutex;
std::condition_variable condition;
bool ready = false;

void consumer()
{
    std::unique_lock<std::mutex> lock(mutex);

    condition.wait(lock, [] {
        return ready;
    });

    use_data();
}

生产者修改状态并通知消费者：

void producer()
{
    {
        std::lock_guard<std::mutex> lock(mutex);
        ready = true;
    }

    condition.notify_one();
}

条件变量的核心作用是：

当条件不满足时让线程休眠；当状态可能发生变化时唤醒线程重新检查条件。

2. 基本使用方法

2.1 三个核心对象

一个完整的条件变量协议通常包含三个部分：

std::mutex mutex;
std::condition_variable condition;
bool ready = false;
共享状态
bool ready;

它表示线程真正关心的业务条件。

常见条件包括：

ready
finished
count > 0
!queue.empty()
stopRequested || !tasks.empty()
state == State::Ready

条件变量本身不是条件，也不会保存业务状态。

互斥锁
std::mutex mutex;

互斥锁用于保护共享状态，保证：

  • 检查条件时不会发生数据竞争；
  • 修改条件时不会发生数据竞争；
  • “检查状态”和“进入等待”能够正确配合。
条件变量
std::condition_variable condition;

条件变量负责：

  • 挂起等待线程；
  • 接收通知；
  • 唤醒一个或多个等待线程。

2.2 等待线程的标准写法

void consumer()
{
    std::unique_lock<std::mutex> lock(mutex);

    condition.wait(lock, [] {
        return ready;
    });

    use_data();
}

如果条件是成员变量：

class Worker {
public:
    void wait_until_ready()
    {
        std::unique_lock<std::mutex> lock(mutex_);
    
        condition_.wait(lock, [this] {
            return ready_;
        });
    }
private:
    std::mutex mutex_;
    std::condition_variable condition_;
    bool ready_{false};
};

2.3 通知线程的标准写法

void producer()
{
    {
        std::lock_guard<std::mutex> lock(mutex);
        ready = true;
    }

    condition.notify_one();
}

推荐的执行顺序是：

获得互斥锁
    ↓
修改共享状态
    ↓
释放互斥锁
    ↓
发送通知

2.4 为什么通知通常放在锁外

下面的写法是正确的：

{
    std::lock_guard<std::mutex> lock(mutex);
    ready = true;
}

condition.notify_one();

被唤醒的线程需要重新获取 mutex 才能从 wait() 返回。

如果在持锁时通知：

{
    std::lock_guard<std::mutex> lock(mutex);

    ready = true;
    condition.notify_one();
}

等待线程虽然被唤醒，但通常仍然要等待通知线程释放 mutex。

因此，在状态已经安全修改的前提下，通常先释放锁再通知。

但不能颠倒为：

condition.notify_one();

{
    std::lock_guard<std::mutex> lock(mutex);
    ready = true;
}

等待线程可能先醒来，发现条件仍然不成立，然后重新等待；生产者修改状态后又没有再次通知，线程可能永久休眠。

3. wait() 的工作原理

3.1 wait() 的三个步骤

下面的代码：

std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock);

在概念上执行：

1. 原子地释放 mutex，并进入等待状态
2. 等待其他线程发出通知
3. 被唤醒后重新获取 mutex

只有重新获得互斥锁后，wait() 才会返回。

因此：

condition.wait(lock);

返回时，lock 仍然持有 mutex。

3.2 为什么释放锁和进入等待必须是原子的

假设手动实现：

lock.unlock();

// 准备进入等待
sleep();

可能发生：

消费者释放锁
    ↓
生产者修改状态并通知
    ↓
此时消费者尚未进入等待，通知被错过
    ↓
消费者开始等待
    ↓
如果不再有通知，消费者永久休眠

condition_variable::wait() 把“解锁”和“进入等待”组合成一个正确的等待操作，从而避免这个竞态窗口。

3.3 为什么必须使用 unique_lock

std::condition_variable::wait() 要求参数为：

std::unique_lock<std::mutex>&

因为等待过程中需要：

lock.unlock();
lock.lock();

std::lock_guard 不支持手动解锁和重新加锁，所以不能用于 wait()：

std::lock_guard<std::mutex> lock(mutex);

// 编译错误
condition.wait(lock);

等待线程应使用：

std::unique_lock<std::mutex> lock(mutex);

只修改状态的通知线程仍然可以使用：

std::lock_guard<std::mutex> lock(mutex);

4. 为什么必须使用谓词

4.1 错误写法

condition.wait(lock);
use_data();

这段代码错误地假设：

wait() 返回就意味着业务条件已经成立。

条件变量只能说明状态可能发生了变化，不能保证条件成立。

4.2 循环检查

正确方式是：

while (!ready) {
    condition.wait(lock);
}

更推荐使用谓词版本：

condition.wait(lock, [] {
    return ready;
});

谓词版本在逻辑上大致等价于：

while (!ready) {
    condition.wait(lock);
}

4.3 虚假唤醒

线程可能在没有对应业务通知的情况下从 wait() 返回，这称为虚假唤醒。

因此：

condition.wait(lock);

返回后，ready 仍然可能是 false。

使用谓词可以自动重新等待：

condition.wait(lock, [] {
    return ready;
});

4.4 多个消费者之间的竞争

假设队列中只加入一个元素，但生产者使用：

condition.notify_all();

两个消费者同时醒来：

消费者 A 获得 mutex
    ↓
A 取走唯一元素
    ↓
A 释放 mutex
    ↓
消费者 B 获得 mutex
    ↓
此时队列已经为空

如果 B 不重新检查条件，就可能访问空队列。

正确写法：

condition.wait(lock, [this] {
    return !queue_.empty();
});

B 发现队列已经为空，会重新等待。

4.5 条件变量不保存历史通知

条件变量不是信号计数器。

如果没有线程等待：

condition.notify_one();

这次通知通常不会被保存。

稍后线程执行：

condition.wait(lock);

不会因为之前发生过通知而自动返回。

所以程序不能依赖“通知是否发生过”，必须依赖共享状态：

condition.wait(lock, [] {
    return ready;
});

即使通知发生在等待之前，只要：

ready == true

等待线程就不会真正进入休眠。

5. 通知接口

5.1 notify_one()

condition.notify_one();

唤醒一个等待线程。

适合一次状态变化只满足一个等待者的场景，例如：

  • 队列新增一个任务；
  • 新增一个可用连接；
  • 释放一个资源；
  • 线程池提交一个任务。
{
    std::lock_guard<std::mutex> lock(mutex);
    tasks.push(std::move(task));
}

condition.notify_one();

5.2 notify_all()

condition.notify_all();

唤醒所有等待线程。

适合：

  • 系统关闭；
  • 全局状态发生变化；
  • 所有线程都需要重新检查退出条件；
  • 新状态可能允许多个线程同时继续。
{
    std::lock_guard<std::mutex> lock(mutex);
    stopping = true;
}

condition.notify_all();

5.3 惊群效应

如果大量线程等待同一个条件：

condition.notify_all();

会导致所有线程同时醒来并争抢同一个 mutex。

但最终可能只有一个线程能取得资源，其余线程又重新进入等待，这称为惊群效应。

因此：

  • 一个资源到达时，优先使用 notify_one()；
  • 全局停止或多个等待者都可能继续时，使用 notify_all()。

6. 超时等待

6.1 wait_for()

等待指定时长：

using namespace std::chrono_literals;

std::unique_lock<std::mutex> lock(mutex);

bool success = condition.wait_for(
    lock,
    500ms,
    [] {
        return ready;
    }
);

返回：

true   // 谓词成立
false  // 超时后谓词仍不成立

使用示例：

if (condition.wait_for(
        lock,
        std::chrono::seconds(1),
        [] {
            return ready;
        })) {
    use_data();
} else {
    handle_timeout();
}

6.2 不带谓词的 wait_for()

std::cv_status status =
    condition.wait_for(lock, 500ms);

返回：

std::cv_status::no_timeout
std::cv_status::timeout

即使返回 no_timeout，也不意味着业务条件一定成立，因此仍然需要检查状态。

6.3 wait_until()

等待到某个时间点：

auto deadline =
    std::chrono::steady_clock::now() +
    std::chrono::seconds(1);

bool success = condition.wait_until(
    lock,
    deadline,
    [] {
        return ready;
    }
);

涉及超时时通常优先使用：

std::chrono::steady_clock

它是单调时钟，不受用户修改系统时间或网络校时影响。

6.4 虚假唤醒导致超时累计

错误写法：

while (!ready) {
    if (condition.wait_for(lock, 1s) ==
        std::cv_status::timeout) {
        return false;
    }
}

如果每隔一段时间发生一次虚假唤醒，每次循环都会重新等待一秒，总时间可能远超一秒。

更稳定的方式是计算一次截止时间：

auto deadline =
    std::chrono::steady_clock::now() + 1s;

while (!ready) {
    if (condition.wait_until(lock, deadline) ==
        std::cv_status::timeout) {
        return ready;
    }
}

或者直接使用带谓词的版本：

bool success =
    condition.wait_for(lock, 1s, [] {
        return ready;
    });

7. 典型应用：阻塞队列

7.1 基础实现

#include <condition_variable>
#include <mutex>
#include <queue>
#include <utility>

template<class T>
class BlockingQueue {
public:
    void push(T value)
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push(std::move(value));
        }

        condition_.notify_one();
    }

    T pop()
    {
        std::unique_lock<std::mutex> lock(mutex_);

        condition_.wait(lock, [this] {
            return !queue_.empty();
        });

        T value = std::move(queue_.front());
        queue_.pop();

        return value;
    }

private:
    std::mutex mutex_;
    std::condition_variable condition_;
    std::queue<T> queue_;
};

执行流程：

消费者调用 pop
    ↓
队列为空，消费者休眠
    ↓
生产者调用 push
    ↓
生产者加入元素并通知
    ↓
消费者醒来并重新获得 mutex
    ↓
消费者取出元素

7.2 为什么取出元素时仍要持锁

下面的写法不安全：

condition_.wait(lock, [this] {
    return !queue_.empty();
});

lock.unlock();

T value = std::move(queue_.front());
queue_.pop();

解锁后，其他线程可能先一步修改队列。

正确做法是在持锁状态下完成：

T value = std::move(queue_.front());
queue_.pop();

取出数据后再释放锁。

8. 支持关闭的阻塞队列

8.1 为什么需要关闭状态

如果消费者只等待：

!queue_.empty()

系统关闭时，空队列上的消费者可能永久等待。

因此条件应包含：

closed_ || !queue_.empty()

8.2 完整实现

#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>
#include <utility>

template<class T>
class BlockingQueue {
public:
    bool push(T value)
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);

            if (closed_) {
                return false;
            }

            queue_.push(std::move(value));
        }

        notEmpty_.notify_one();
        return true;
    }

    std::optional<T> pop()
    {
        std::unique_lock<std::mutex> lock(mutex_);

        notEmpty_.wait(lock, [this] {
            return closed_ || !queue_.empty();
        });

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

        T value = std::move(queue_.front());
        queue_.pop();

        return value;
    }

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

        notEmpty_.notify_all();
    }

private:
    std::mutex mutex_;
    std::condition_variable notEmpty_;
    std::queue<T> queue_;
    bool closed_{false};
};

8.3 消费者使用方式

BlockingQueue<int> queue;

void consumer()
{
    while (auto value = queue.pop()) {
        process(*value);
    }

    // 队列已关闭并且没有剩余数据
}

关闭：

queue.close();

close() 必须使用：

notify_all()

因为所有消费者都需要醒来并退出。

9. 有界阻塞队列

9.1 为什么需要两个条件变量

固定容量队列存在两个条件：

  • 消费者等待队列非空；
  • 生产者等待队列未满。

因此通常使用：

std::condition_variable notEmpty_;
std::condition_variable notFull_;

9.2 实现

template<class T>
class BoundedQueue {
public:
    explicit BoundedQueue(std::size_t capacity)
        : capacity_(capacity)
    {
    }

    void push(T value)
    {
        std::unique_lock<std::mutex> lock(mutex_);

        notFull_.wait(lock, [this] {
            return queue_.size() < capacity_;
        });

        queue_.push(std::move(value));

        lock.unlock();
        notEmpty_.notify_one();
    }

    T pop()
    {
        std::unique_lock<std::mutex> lock(mutex_);

        notEmpty_.wait(lock, [this] {
            return !queue_.empty();
        });

        T value = std::move(queue_.front());
        queue_.pop();

        lock.unlock();
        notFull_.notify_one();

        return value;
    }

private:
    const std::size_t capacity_;

    std::queue<T> queue_;
    std::mutex mutex_;

    std::condition_variable notEmpty_;
    std::condition_variable notFull_;
};

状态转换关系：

push：
等待队列未满
    ↓
放入元素
    ↓
通知队列非空

pop：
等待队列非空
    ↓
移除元素
    ↓
通知队列未满

10. 在线程池中的应用

10.1 工作线程

void ThreadPool::worker_loop()
{
    while (true) {
        Task task;

        {
            std::unique_lock<std::mutex> lock(mutex_);

            condition_.wait(lock, [this] {
                return stopping_ || !tasks_.empty();
            });

            if (stopping_ && tasks_.empty()) {
                return;
            }

            task = std::move(tasks_.front());
            tasks_.pop();
        }

        task();
    }
}

谓词：

stopping_ || !tasks_.empty()

表示工作线程在以下情况醒来：

  • 队列中有任务；
  • 线程池正在停止。

10.2 提交任务

{
    std::lock_guard<std::mutex> lock(mutex_);

    if (stopping_) {
        throw std::runtime_error(
            "thread pool has stopped"
        );
    }

    tasks_.push(std::move(task));
}

condition_.notify_one();

新增一个任务通常只需要唤醒一个工作线程。

10.3 关闭线程池

{
    std::lock_guard<std::mutex> lock(mutex_);
    stopping_ = true;
}

condition_.notify_all();

所有工作线程都需要醒来检查退出条件。

10.4 为什么任务必须在锁外执行

错误写法：

std::unique_lock<std::mutex> lock(mutex_);

condition_.wait(lock, [this] {
    return !tasks_.empty();
});

Task task = std::move(tasks_.front());
tasks_.pop();

task(); // 仍然持有 mutex

这样会导致：

  • 同一时间只能有一个工作线程执行任务；
  • 长任务阻止其他线程取任务；
  • 任务内部再次提交任务时可能死锁；
  • 线程池实际上被串行化。

正确方式：

Task task;

{
    std::unique_lock<std::mutex> lock(mutex_);

    condition_.wait(lock, [this] {
        return stopping_ || !tasks_.empty();
    });

    task = std::move(tasks_.front());
    tasks_.pop();
}

task();

11. 内存可见性

11.1 可见性来自互斥锁

生产者：

{
    std::lock_guard<std::mutex> lock(mutex);
    data = 42;
    ready = true;
}

condition.notify_one();

消费者：

std::unique_lock<std::mutex> lock(mutex);

condition.wait(lock, [] {
    return ready;
});

std::cout << data;

消费者能够安全看到：

data == 42

原因不是通知本身“传递了数据”，而是：

生产者释放 mutex
    synchronizes-with
消费者重新获取同一个 mutex

由此建立 happens-before 关系。

11.2 通知不是内存屏障的替代品

错误写法：

data = 42;
ready = true;
condition.notify_one();

如果其他线程使用 mutex 访问 data 和 ready，生产者也必须遵守同一个同步协议。

正确性来自：

共享状态 + 同一个 mutex + 条件变量

而不是单独来自 notify_one()。

12. condition_variable_any

12.1 与普通条件变量的区别

std::condition_variable 直接配合：

std::unique_lock<std::mutex>

std::condition_variable_any 可以配合其他满足锁要求的类型：

std::condition_variable_any condition;
std::shared_mutex mutex;

std::unique_lock<std::shared_mutex> lock(mutex);

condition.wait(lock, [] {
    return ready;
});

它也可以配合自定义锁。

12.2 选择建议

如果使用普通的：

std::mutex

优先选择：

std::condition_variable

只有确实需要其他锁类型时，才选择：

std::condition_variable_any

后者更通用，但实现可能有额外开销。

13. C++20 原子等待

C++20 为原子变量增加了：

atomic.wait()
atomic.notify_one()
atomic.notify_all()

简单标志可以写成：

std::atomic<bool> ready{false};

void consumer()
{
    ready.wait(false);

    use_data();
}

void producer()
{
    prepare_data();

    ready.store(
        true,
        std::memory_order_release
    );

    ready.notify_one();
}

消费者可以配合 acquire：

while (!ready.load(std::memory_order_acquire)) {
    ready.wait(false);
}

适用场景对比：

场景 推荐工具
等待一个简单原子值变化 atomic::wait
等待复合条件 condition_variable
等待队列状态 condition_variable
等待停止或任务到达 condition_variable
需要多个状态共同判断 condition_variable

例如：

stopping || !tasks.empty()

这种复合条件更适合条件变量。

14. 常见错误

14.1 不使用谓词

错误：

condition.wait(lock);
consume();

正确：

condition.wait(lock, [] {
    return ready;
});

14.2 使用 if 而不是循环

错误：

if (!ready) {
    condition.wait(lock);
}

正确：

while (!ready) {
    condition.wait(lock);
}

或：

condition.wait(lock, [] {
    return ready;
});

14.3 修改共享状态时没有加锁

错误：

ready = true;
condition.notify_one();

正确：

{
    std::lock_guard<std::mutex> lock(mutex);
    ready = true;
}

condition.notify_one();

14.4 使用不同的 mutex

如果等待线程使用：

mutexA

通知线程修改状态时却使用：

mutexB

就没有形成统一同步协议。

检查和修改同一个条件状态时，应使用同一个 mutex。

14.5 把通知理解成条件成立

错误理解：

收到通知 = 条件已经成立

正确理解：

收到通知 = 条件可能发生变化，必须重新检查

14.6 按值捕获谓词状态

错误：

condition.wait(lock, [ready] {
    return ready;
});

这里检查的是 ready 的旧副本。

正确：

condition.wait(lock, [&ready] {
    return ready;
});

成员变量：

condition.wait(lock, [this] {
    return ready_;
});

14.7 关闭时只通知一个线程

错误：

stopping = true;
condition.notify_one();

其他等待线程可能永远无法退出。

正确：

{
    std::lock_guard<std::mutex> lock(mutex);
    stopping = true;
}

condition.notify_all();

14.8 在持锁状态下调用外部代码

std::unique_lock<std::mutex> lock(mutex);

condition.wait(lock, [] {
    return ready;
});

callback(); // 仍然持有 mutex

callback() 可能：

  • 执行很久；
  • 再次获取同一把锁；
  • 获取其他锁；
  • 等待 I/O；
  • 调用未知代码。

应在锁内取得必要数据，然后解锁：

auto data = sharedData;

lock.unlock();

callback(data);

14.9 在持锁状态下等待其他线程

std::lock_guard<std::mutex> lock(mutex);
worker.join();

如果 worker 退出前需要获取 mutex，就会死锁。

类似地，应谨慎在持锁状态下调用：

future.get();
thread.join();
sleep();
blocking_io();

15. 标准使用模板

15.1 等待模板

std::unique_lock<std::mutex> lock(mutex);

condition.wait(lock, [&] {
    return stopRequested ||
           condition_is_satisfied();
});

if (stopRequested &&
    !condition_is_satisfied()) {
    return;
}

// 在锁保护下取得或更新共享状态

15.2 通知模板

{
    std::lock_guard<std::mutex> lock(mutex);
    update_shared_state();
}

condition.notify_one();

15.3 关闭模板

{
    std::lock_guard<std::mutex> lock(mutex);
    stopRequested = true;
}

condition.notify_all();

15.4 超时模板

std::unique_lock<std::mutex> lock(mutex);

bool success = condition.wait_for(
    lock,
    std::chrono::seconds(1),
    [&] {
        return stopRequested ||
               condition_is_satisfied();
    }
);

if (!success) {
    handle_timeout();
}

16. 实用检查清单

使用条件变量时，可以检查以下问题：

  • 业务条件是否由明确的共享状态表示？
  • 共享状态的所有访问是否由同一个 mutex 保护？
  • 等待线程是否使用 std::unique_lock？
  • wait() 是否使用了谓词或 while 循环？
  • 谓词是否检查了停止条件？
  • 修改状态是否发生在通知之前？
  • 一个资源是否使用 notify_one()？
  • 全局关闭是否使用 notify_all()？
  • 从队列中取出数据时是否仍然持有锁？
  • 耗时任务是否在释放锁之后执行？
  • 超时逻辑是否考虑虚假唤醒？
  • 等待对象、mutex 和条件变量的生命周期是否足够长？
  • 是否错误地把超时理解成取消？
  • 是否可能在持锁状态下调用 join() 或 future::get()？

17. 总结

条件变量不是单独工作的，而是一个完整协议：

共享状态
    +
mutex
    +
condition_variable

最重要的规则是：

  1. 条件变量不是条件，共享状态才是条件。
  2. 共享状态必须由 mutex 保护。
  3. wait() 会释放锁，醒来后重新获取锁。
  4. 必须使用谓词或循环检查条件。
  5. 通知只表示状态可能变化。
  6. 条件变量不会保存历史通知。
  7. 一个资源通常使用 notify_one()。
  8. 全局停止通常使用 notify_all()。
  9. 应在锁内取得任务，在锁外执行任务。
  10. 超时只表示不再等待，不代表操作被取消。

最推荐记住的两段代码是：

std::unique_lock<std::mutex> lock(mutex);

condition.wait(lock, [&] {
    return stopRequested || !queue.empty();
});

以及：

{
    std::lock_guard<std::mutex> lock(mutex);
    update_shared_state();
}

condition.notify_one();

只要始终围绕“先检查共享状态，而不是相信通知”来设计，条件变量就会清晰很多。