C++ condition_variable详解：从wait/notify原理到生产者消费者实战

C并发系列写到第四篇终于轮到 condition_variable 这个在生产者消费者、任务队列、线程池里出镜率极高的同步原语。前面几篇我们把线程创建、mutex 互斥、原子操作都过了一遍但很多人写完这些还是会卡壳锁能保证“同一时间只有一个人进”却没法解决“我什么时候可以干活”这种问题。你总不能让线程死循环去抢锁吧那 CPU 早就烧没了。condition_variable 就是来解决这个“等待-通知”问题的。这篇文章我尽量不绕弯子先把机制讲透再给一个可以完整编译运行的案例最后用 Java 的 wait/notify、Lock 和 BlockingQueue 做对照帮你在面试和实际开发里都能把两边的知识串起来。1. 从本质讲起condition_variable 到底解决了什么问题1.1 没有通知机制的世界轮询与延迟陷阱先看一个非常常见的场景。假设有一个全局的任务队列生产线程往里塞任务消费线程从里面取任务。在没有 condition_variable 之前最原始的写法是让消费线程不停地循环std::mutex mtx; std::dequeint tasks; void consumer() { while (true) { std::lock_guardstd::mutex lock(mtx); if (!tasks.empty()) { int task tasks.front(); tasks.pop_front(); // 处理任务 } } }这个代码逻辑上没错但有两个无法接受的问题。第一CPU 会被这个空转的循环占满一个消费者线程就能把一个核烧到 100%而且它什么都没干纯粹在检查队列。第二如果你为了让它在没任务时休息一下在空队列分支里加一个std::this_thread::sleep_for(10ms)那任务的延迟就完全取决于这个 sleep 的间隔。想去掉延迟就得缩短 sleep想省 CPU 就得加长 sleep两头都难受。我在实际项目里见过类似的代码一个消费者线程占满了一整颗核排查了半天才发现是这种轮询写法改成条件变量之后 CPU 直接降到接近 0。1.2 条件变量解决的两个核心问题condition_variable 要做的事情其实就两件。第一把“检查条件”和“进入等待”变成一个不可分割的整体。如果先检查条件再等待中间可能被别人插一脚导致通知丢在“检查完”和“开始等”的缝隙里这也就是所谓的 lost wakeup丢失唤醒。条件变量通过和 mutex 配合把这两个动作绑定到 wait 函数内部从根源上堵住了这个缝隙。第二等待的时候真正挂起线程让出 CPU直到别人通知它再醒来。这就像你去餐厅吃饭没有叫号系统的时候你只能每隔几分钟跑去问服务员“有空位了吗”既累又烦有了叫号系统你可以在等候区安心休息服务员喊到你的时候再起身进去。mutex 是门锁condition_variable 就是那个叫号器两者协同工作数据本身由 mutex 保护数据状态的变化由 condition_variable 通知。2. 核心 API 与机制剖析wait / notify 到底怎么配合2.1 wait 系列自动释放锁与被唤醒后的锁回先看最简单的wait(lock)这个接口的调用前提是当前线程已经持有了传入的锁。进入 wait 之后它会原子地完成两件事把当前线程放进等待队列然后释放掉这个锁。为什么要释放因为当前线程要睡觉了不可能攥着锁不放手否则其他线程无法修改共享数据也就没人能来叫醒它。当收到 notify 通知并且当前线程成功抢到锁之后wait 才会返回。这里有一个特别重要的点wait 返回的时候条件并不一定成立。可能有两个消费者同时被唤醒其中一个先抢到锁把队列里的任务取走了另一个抢到锁后发现队列又空了。所以标准的用法永远是循环检查条件而不是用 ifstd::unique_lockstd::mutex lock(mtx); while (tasks.empty()) { cv.wait(lock); } // 到这里 tasks 一定不为空这个 while 循环写多了之后C 直接在 overload 版本里帮你封装了它。下面这两种写法是完全等价的// 写法一手动 while wait while (tasks.empty()) { cv.wait(lock); } // 写法二带 predicate 的 wait cv.wait(lock, []() { return !tasks.empty(); });带 predicate 的版本是我日常用得最多的因为它把“条件不满足就继续等”这个意图表达得非常清楚也避免了忘记写 while 的低级错误。它内部就是 while (!pred()) wait(lock) 的简写。超时接口需要单独说。wait_for和wait_until都有两个版本带 predicate 和不带 predicate 的// 不带 predicate 版本返回 std::cv_status std::cv_status status cv.wait_for(lock, std::chrono::milliseconds(100)); if (status std::cv_status::timeout) { // 超时 } else { // 被唤醒 } // 带 predicate 版本返回 bool bool ready cv.wait_for(lock, std::chrono::milliseconds(100), []() { return flag; }); if (ready) { // 条件成立 } else { // 超时 }带 predicate 的版本返回值就是条件是否成立用起来最省心。手动版本要注意返回timeout不能等同于“条件不成立”因为完全有可能在超时的那一瞬间恰好有人 notify这时候你去读条件可能已经成立了。所以无论哪种写法超时返回之后都必须再检查一次实际条件这是很多 bug 的源头。2.2 notify_one 与 notify_all 的选型和细节notify_one负责唤醒等待队列里的一个线程notify_all负责唤醒所有线程。选型不能只看等待线程数量更要看这次通知之后被唤醒的线程是不是都能继续干活。如果队列里只有一条任务你唤醒所有消费者它们会同时醒来抢锁最后只有一个能拿到任务其他几个只能继续回去睡。这个现象有点像惊群效应白白增加调度开销。所以单任务场景用notify_one就够了。反过来如果是广播型事件比如服务器要关闭了、任务队列要清空了这种所有等待线程都需要感知的状态变化必须用notify_all否则只唤醒一个其他线程永远不知道发生了改变。还有一个经验是 notify 的时机问题。调用 notify 的时候不一定非要持有锁相反我建议先把锁释放掉再通知。我给你看一个细节对比// 持锁通知 { std::lock_guardstd::mutex lock(mtx); tasks.push_back(1); cv.notify_one(); // 被唤醒的线程想去抢锁但锁还没释放 } // 释放后再通知 { std::unique_lockstd::mutex lock(mtx); tasks.push_back(1); lock.unlock(); cv.notify_one(); // 被唤醒的线程一醒来就能拿到锁 }第二种写法能让被唤醒的线程立刻拿到锁减少“唤醒了又阻塞在锁上”的来回抖动。有些场景里这个细节对吞吐量有明显影响后面工程实践部分我再细说。2.3 为什么 wait 必须和同一把 mutex 配合条件变量和 mutex 的关系经常让人困惑我当年也迷过一阵。其实核心是为了保证“检查-等待”的原子性。设想一种错误的写法std::mutex mtx; std::condition_variable cv; bool ready false; // 线程 A生产者 { std::lock_guardstd::mutex lock(mtx); ready true; } cv.notify_one(); // 线程 B消费者 if (!ready) { // 在锁外检查 cv.wait(lock); // 这里会一直等下去 }问题出在线程 B 检查 ready 和调用 wait 之间不是原子的。如果线程 A 在 B 检查完 ready此时还是 false之后、B 调用 wait 之前把 ready 改成 true 并且调用了 notify那么这次通知就发生在 B 正式进入等待之前直接丢了。从此 B 就永远等在那里即使 ready 已经是 true。正确做法是让条件检查和 wait 在同一个锁的保护下并且直接使用带 predicate 的重载。因为 wait 内部会在持有锁的情况下检查 predicate不满足才原子地释放锁并进入等待这样线程 A 即使先改了状态再通知B 进入 wait 时也会先看一眼 predicate发现条件已经满足就不会真的睡过去。这就是 condition_variable 为什么必须和同一把 mutex 绑在一起的根本原因。3. 完整案例用 condition_variable 实现一个带缓冲的任务队列3.1 场景与设计哪些条件需要等待这次写一个经典但足够完整的例子一个容量有限的任务队列多个生产者往里放任务多个消费者从里面取任务。这个队列有两个需要等待的场景队列为空时消费者不能取需要等“队列非空”的条件。队列满时生产者不能放需要等“队列有空位”的条件。所以最自然的设计是使用两个 condition_variable一个管“非空”一个管“非满”。Java 的ArrayBlockingQueue内部就是这种双 Condition 结构我们这里用 C 手写一遍你会发现两边几乎是一一对应的。3.2 完整代码与逐段解释完整代码如下可以直接编译运行#include condition_variable #include deque #include iostream #include mutex #include thread #include chrono class BlockingQueue { public: explicit BlockingQueue(size_t maxSize) : maxSize_(maxSize) {} void push(int val) { std::unique_lockstd::mutex lock(mtx_); notFull_.wait(lock, []() { return queue_.size() maxSize_; }); queue_.push_back(val); std::cout push val , queue size queue_.size() std::endl; // 释放锁之后再通知消费者减少锁竞争 lock.unlock(); notEmpty_.notify_one(); } int pop() { std::unique_lockstd::mutex lock(mtx_); notEmpty_.wait(lock, []() { return !queue_.empty(); }); int val queue_.front(); queue_.pop_front(); std::cout pop val , queue size queue_.size() std::endl; lock.unlock(); notFull_.notify_one(); return val; } private: std::mutex mtx_; std::condition_variable notEmpty_; std::condition_variable notFull_; std::dequeint queue_; size_t maxSize_; }; int main() { BlockingQueue queue(2); std::thread consumer1([]() { for (int i 0; i 5; i) { queue.pop(); std::this_thread::sleep_for(std::chrono::milliseconds(20)); } }); std::thread consumer2([]() { for (int i 0; i 5; i) { queue.pop(); std::this_thread::sleep_for(std::chrono::milliseconds(30)); } }); for (int i 0; i 10; i) { queue.push(i); std::this_thread::sleep_for(std::chrono::milliseconds(50)); } consumer1.join(); consumer2.join(); return 0; }逐个拆解关键段落。push里第一步是加锁然后调用notFull_.wait(lock, ...)。这个 wait 有两个作用如果队列已经满了当前生产线程会阻塞在这里直到消费者取走数据后调用notFull_.notify_one()如果队列没满wait 会立即返回继续往下执行。wait 内部在阻塞期间会自动释放锁这使得消费者在队列满时依然能够进入pop并取走数据。入队之后我特意先lock.unlock()再notEmpty_.notify_one()。这么做的好处前面说过被唤醒的消费者可以立刻获得锁不必等待生产者在作用域末尾释放。注意unique_lock不像lock_guard那样析构时才解锁它允许你手动控制解锁时机这就是我在这里用它的原因。pop的逻辑是对称的。消费者在notEmpty_.wait上等待队列为空时挂起弹出数据后手动解锁然后notFull_.notify_one()唤醒一个等待中的生产者。3.3 运行结果观察与参数调整建议正常运行时会看到生产者输出几条 “push” 之后消费者开始输出 “pop”期间队列大小在 0 到 2 之间波动。因为消费线程启动后就会立刻尝试pop而队列初始是空的所以两个消费者都会先阻塞在notEmpty_.wait。生产者每隔 50ms 推入一条消费者则按照各自的节奏取走到了第 5、6 条左右因为 maxSize 是 2生产者可能会被notFull_.wait卡住直到消费者取走数据腾出空位。我建议动手改几个参数观察行为变化。比如把 maxSize 改成 1整个队列就退化为一个“槽位”生产者和消费者必须严格交替执行你能看到非常清晰的阻塞-唤醒过程。再比如把消费者从这个改成 3 个或者更多看看notify_one是否会导致某些消费者长期得不到任务。这种微调比看任何理论讲解都更能理解条件变量的行为。这里有一个小提醒代码里的std::cout本身不是线程安全的但这个示例里每个输出都发生在队列锁释放之前所以从共享数据上讲是安全的。实际项目中如果日志系统比较复杂建议给日志单独加锁或者用线程安全的日志库不要顺手往业务锁里塞日志输出。4. Java 对比视角从 Object.wait 到 Lock 再到 BlockingQueue4.1 Object.wait/notify 与 condition_variable 的等价物Java 里每个对象都可以作为锁和等待集合这是和 C 一个很大的思维差异。synchronized(lock)代码块里你可以调用lock.wait()让当前线程释放 monitor 并挂起其他线程持锁时调用lock.notify()唤醒一个等待者。逻辑上这几乎就是 condition_variable 的 Object 内建版本// 消费者 synchronized (lock) { while (queue.isEmpty()) { lock.wait(); } int val queue.removeFirst(); } // 生产者 synchronized (lock) { queue.addLast(val); lock.notify(); }注意 Java 里wait()必须在synchronized块内调用这对应 C 的“wait 时当前线程必须持有锁”。Java 也必须使用 while 循环重新检查条件原因和 C 完全一样虚假唤醒和竞争唤醒在 Java 里一样存在。最大的差异是 Java 的wait()声明会抛出InterruptedException所以要么在方法签名里加上throws要么用 try/catch 包起来。C 的线程模型没有这种中断机制这是两种语言设计取向的不同不是简单的谁好谁坏。4.2 Condition 接口结构和 API 几乎一致的对应如果你用过 Java 的ReentrantLock会发现它提供的Condition接口和 C 的 condition_variable 在结构上几乎一一对应ReentrantLock lock new ReentrantLock(); Condition notEmpty lock.newCondition(); Condition notFull lock.newCondition(); // 生产者 lock.lock(); try { while (count items.length) { notFull.await(); } // 写入数组 notEmpty.signal(); } finally { lock.unlock(); }对应的 C 结构std::mutex mtx; std::condition_variable notEmpty; std::condition_variable notFull; // 生产者 std::unique_lockstd::mutex lock(mtx); notFull.wait(lock, []() { return count items.length; }); // 写入数据 notEmpty.notify_one();这里面的对应关系非常清晰wait对应awaitnotify_one对应signalnotify_all对应signalAll。Condition的优势在于它允许你在同一个锁上创建多个独立的等待集合这正好对应 C 里用多个 condition_variable 配合一个 mutex 的做法。我建议用一张表把这些对应关系记下来面试被问到“C 和 Java 的并发原语怎么对应”时可以直接拿出来用功能CJava ObjectJava Condition锁std::mutex / unique_locksynchronizedReentrantLock等待集合对象condition_variable对象自身Condition等待wait(lock, pred)wait()await()唤醒单个notify_one()notify()signal()唤醒全部notify_all()notifyAll()signalAll()超时等待wait_for / wait_untilwait(timeout)await(timeout)中断支持无InterruptedExceptionInterruptedException4.3 BlockingQueueJava 工程中的封装替代实际写 Java 生产代码的时候我几乎不会手写 Condition 去做生产者消费者直接用ArrayBlockingQueue或者LinkedBlockingQueue就结束了。这不是因为 Java 开发者比 C 开发者懒而是 JDK 已经把“wait while signal”这套底层逻辑封装好了还考虑到了公平性、超时、中断等一堆细节。ArrayBlockingQueue内部正是用两个 ConditionnotEmpty和notFull实现的和上面 C 版本的思路完全一样只是平时你不用自己写而已。C 标准库里没有等价的阻塞队列容器所以我们要自己封装。这算是语言生态上的差异不是能力上的差异。理解这个背景之后你在面试时就可以这样回答“Java 的 BlockingQueue 提高了并发编程的上层抽象C 则把这些控制权留给了开发者各有取舍。”这个回答既展示了底层的理解又体现了对大厂封装程度的认知。4.4 两边在工程思维上的取舍对比除了 API 层面的差别两边在工程思维上也有值得注意的差异。Java 的ReentrantLock支持公平锁、可中断锁等待、多个 Condition 绑定同一个锁语言层面提供了更多托管运行时的便利。C 则讲究零开销抽象没有 GC 兜底也没有语言级别的中断所以写 C 并发时对资源管理和生命周期要更敏感。比如 C 里 condition_variable 不能拷贝要小心和对象生命周期绑定Java 里这些内存管理问题被 JVM 接管了。不过在这些底层语义上两者遵循的是同一套并发理论。虚假唤醒、丢失唤醒、线程竞争这些概念在两边都存在解决问题的思路也一致条件检查必须用 while条件状态的变化必须由锁保护唤醒必须发生在状态变化之后。能把 C 的机制理解透彻再去看 Java 的封装几乎是一马平川反过来从 Java 的高层抽象出发也能帮你理解 C 底层为什么要提供这些原语。5. 工程实践中常见的坑与排查技巧实录5.1 丢失唤醒最隐蔽也最致命的坑丢失唤醒大概是条件变量领域最臭名昭著的问题。它难排查因为它不是每次都发生往往取决于线程调度的时序。我在上文讲过如果“检查条件”和“进入等待”不是原子的通知就可能落在两者之间导致等待方永远睡过头。用 predicate 重载能解决这个问题因为 wait 内部把“检查条件 决定是否等待”绑定成了一个原子操作。这里有一个我的切身体会。之前维护一个老项目同事在代码里用条件变量做缓存刷新刷新线程发现缓存过期后不是先置一个标志位再通知而是直接 notify。等待线程醒来后再次检查标志位发现没变又继续睡。这个 bug 在测试环境整整两天才复现一次后来通过打印日志才发现唤醒比状态变更提前了。正确的顺序永远是先修改受锁保护的条件再释放锁最后 notify。顺序错了代码再漂亮也是定时炸弹。5.2 虚假唤醒与超时返回后的二次判断标准库文档明确说 spurious wakeup 是合法的也就是说线程可能在没有任何人调用 notify 的情况下自己醒来。操作系统层面很少见但你不能赌它不发生。C 的 predicate 重载和 while 循环自动帮你处理了这种情况所以只要你坚持用这两种写法虚假唤醒基本不用操心。麻烦的是超时。很多人写超时逻辑时会这样std::unique_lockstd::mutex lock(mtx); if (cv.wait_for(lock, std::chrono::seconds(1)) std::cv_status::timeout) { // 认为条件不成立 } else { // 认为条件成立 }这个写法有隐患。wait_for 返回timeout只能说明超时了不能说明条件一定不成立返回no_timeout也不能保证条件一定成立因为可能是虚假唤醒。最稳妥的写法是带 predicate 的重载bool success cv.wait_for(lock, std::chrono::seconds(1), []() { return flag; }); if (success) { // 条件成立 } else { // 超时或条件始终未成立 }这个版本直接给出“条件是否成立”的结论省去了手动二次判断的麻烦。我用这个 API 之后超时相关的逻辑 bug 少了很多。5.3 持锁 notify 带来的唤醒抖动我在第 2 节提到过持锁 notify 的问题这里展开讲一下。假设你在lock_guard保护的作用域内调用notify_one被唤醒的线程会立刻尝试获取同一把锁。但此时锁还在通知方手里要等lock_guard析构才能释放。于是被唤醒线程刚被唤起来马上又因为抢锁失败而阻塞回去白白消耗一次调度切换。我实际测过一个简化版的任务队列持锁 notify 和解锁后 notify 的吞吐量有百分之几的差别。在队列本身非常短、竞争激烈的时候这个差距会更明显。所以我的习惯是如果条件变量保护的临界区很短就在临界区外通知如果临界区很长这个优化就更值得做。C 的unique_lock可以手动 unlock很灵活Java 里由于 try-finally 释放锁一般就直接在锁内 signal 了这算是工程习惯上的一个小差异。5.4 一次实战排查线程卡死的定位思路最后分享一个我实际排查过的卡死问题。当时一个监控采集服务里用条件变量通知消费者处理过期数据线上出现消费者线程不工作的现象。第一反应是怀疑丢失唤醒于是加日志准备抓时序。用 gdb 挂上进程执行thread apply all bt看所有线程栈发现消费者线程确实阻塞在cv.wait上但生产者线程并没有死掉它正卡在一个网络请求的超时等待里。也就是说条件变量本身没有任何问题是生产者因为外部依赖变慢迟迟没有产生新数据消费者才一直空等。这个案例让我意识到排查条件变量问题时不要只盯着条件变量本身还要把整条数据链路看清楚。gdb 看线程栈是最直接有效的手段其次是在 wait 前后加带时间戳的日志确认到底是“没收到通知”还是“收到了通知但条件不满足”。5.5 我写条件变量时固定检查的三个问题经过前面这些坑我现在每写一段条件变量代码都会在心里过三个问题第一所有对条件状态的读写是否都在同一把锁的保护下第二被唤醒之后是否重新检查了条件而不是直接假设条件成立第三通知方是不是在状态修改完成并且释放锁之后才调用 notify如果三个问题的答案都是肯定的这段代码基本不会再出幺蛾子。另外我还会顺手确认 close/stop 这类广播事件用的是notify_all而不是notify_one否则十有八九会漏掉某个等待线程。从 C 的 condition_variable 到 Java 的 Object.wait 和 Condition再到成熟的 BlockingQueue 封装你会发现并发编程的核心问题其实是相通的怎么让线程在合适的时机睡下又怎么在合适的时机醒来。把 wait/notify 这套机制彻底理解透了后面看什么语言的高并发代码都会顺很多。