★ FEATURED ARTICLE 上一篇文章我们把互斥锁聊透了怎么加锁、怎么避免死锁、锁的粒度怎么控制。但锁解决的只是“互斥”也就是同一时间只有一个人进厕所。现实里的线程协作远不止“你别进来”这么简单更多时候是一个线程要等另一个线程完成任务再开工——这就是“线程同步”的问题。今天这篇继续往下走把条件变量、阻塞队列的生产者消费者模型、环形队列版本、线程池、线程安全和读写锁一口气讲完。前一篇打底的是锁这一篇打底的是“等待与唤醒”两者合在一起才算把多线程协作的骨架搭起来。我默认你已经会写 pthread_mutex_t并且知道 lock/unlock 的基本套路。如果你连互斥锁都还没吃透建议先把上一篇翻完再回来不然今天讲的条件变量和锁的交互关系很容易绕晕。准备好了就开始。1. 从互斥锁到等待唤醒条件变量到底解决什么问题1.1 互斥锁的天然短板只有互斥没有等待先想一个特别常见的场景一个线程往队列里塞数据另一个线程从队列里取数据。如果用互斥锁保护这个队列伪代码长这样while (1) { pthread_mutex_lock(mtx); if (queue_empty()) { pthread_mutex_unlock(mtx); continue; // 空转继续抢锁 } item queue_pop(); pthread_mutex_unlock(mtx); handle(item); }这段代码的问题肉眼可见消费者线程在队列为空的时候会反复抢锁、判断、释放锁完了再抢。这叫忙等待对 CPU 是纯粹的浪费。更恶心的是它还会和生产者抢同一个锁有可能导致生产者拿不到锁数据迟迟塞不进去整个系统在“空转抢夺锁”上消耗大量时间片。有人会说那我加个 sleep 行不行比如空转的时候睡 10 毫秒。能缓解一下但这是拍脑袋定出来的延迟数据到达的节奏和 sleep 根本不是对齐的。数据刚到的时候你可能还在睡数据没到的时候你反而醒着抢锁延迟和吞吐两头不讨好。我们真正需要的是一种机制消费者发现队列为空就老老实实睡过去不占锁也不占 CPU生产者一旦塞入数据能立刻把消费者叫醒。这就是条件变量存在的意义。条件变量解决的问题不是“互斥”而是“等待与通知”。打个比方互斥锁是厕所门锁条件变量是“排号叫号”的系统你不需要一直盯着柜台轮到你时广播会通知你。1.2 条件变量的核心 API 和调用规范Linux 下最常用的条件变量 API 就这么几个pthread_cond_init、pthread_cond_wait、pthread_cond_signal、pthread_cond_broadcast。当然还有带超时版本的后面排查问题时会用到。条件变量本身不包含状态它必须搭配一个互斥锁使用。这是新手最容易懵的地方为什么 wait 要传一把锁进去看这段标准写法pthread_mutex_lock(mtx); while (count 0) { pthread_cond_wait(cond, mtx); } consume(); pthread_mutex_unlock(mtx);pthread_cond_wait做的事要拆成两步理解把mtx释放掉让其他线程能进临界区把自己阻塞在cond上直到被唤醒。注意这两步是原子操作中间不会插入其他线程的调度。为什么必须原子如果不原子就可能出现这种情况你刚判断完count 0还没来得及睡生产者先把数据塞进去并发了 signal然后你才进入睡眠信号就永久丢失了。你会一直睡到天荒地老。所以条件变量的 wait 必须“释放锁 挂起”一气呵成这样才能保证“检查条件到进入睡眠”之间不会有信号漏掉。唤醒后的行为也要注意pthread_cond_wait返回时锁mtx已经被重新拿到手了。所以你会看到 wait 调用通常在锁内返回后也依然在锁内临界区保护没有断。1.3 为什么判断条件必须用 while不能裸用 if我刚学条件变量的时候也犯过这个错直接if (count 0) pthread_cond_wait(...)醒来以后就往下走。结果线上偶尔出 bug查了半天才发现是条件判断的问题。原因有两个。第一个叫虚假唤醒spurious wakeupLinux 的pthread_cond_wait文档明确说了即使是单生产者单消费者的简单场景也可能因为信号处理、调度等原因被唤醒但条件并不满足。第二个原因是多消费者场景你被唤醒后另一个消费者可能已经抢先一步把数据取走了等你拿到锁重新跑起来时条件又变成不满足了。所以规范写法铁律是用 while 循环重新检查条件不满足就继续等while (queue_empty()) { pthread_cond_wait(cond, mtx); }这不是代码洁癖而是正确性问题。任何讲条件变量的教程如果没强调这条都是在埋雷。2. 基于阻塞队列的 CP 模型单条件变量也能跑起来2.1 阻塞队列到底“阻塞”在哪生产者消费者模型里队列两侧的线程节奏往往不一样。数据生产太快、消费太慢队列会堆满生产太慢、消费太快队列会空掉。阻塞队列的思路就是队列空时消费者阻塞队列满时生产者阻塞。队列不再是一个简单被锁保护的容器而是自带“流量控制”语义的同步组件。一个最精简的实现用一个互斥锁和一个条件变量就能完成。我们先不做环形队列那种两个条件的优化先把单条件版本跑通搞清楚状态是怎么传递的。2.2 单条件变量的阻塞队列设计与实现我先写一个 C 版本的简单原型关键是看 push 和 pop 如何应对“空”和“满”两个方向#include pthread.h #include deque templatetypename T class BlockQueue { public: BlockQueue(size_t cap) : capacity_(cap) { pthread_mutex_init(mtx_, nullptr); pthread_cond_init(cond_, nullptr); } ~BlockQueue() { pthread_mutex_destroy(mtx_); pthread_cond_destroy(cond_); } void push(const T item) { pthread_mutex_lock(mtx_); while (queue_.size() capacity_) { pthread_cond_wait(cond_, mtx_); // 队满等消费者腾位置 } queue_.push_back(item); pthread_cond_signal(cond_); // 通知可能等待的消费者 pthread_mutex_unlock(mtx_); } T pop() { pthread_mutex_lock(mtx_); while (queue_.empty()) { pthread_cond_wait(cond_, mtx_); // 队空等生产者投喂 } T item queue_.front(); queue_.pop_front(); pthread_cond_signal(cond_); // 通知可能等待的生产者 pthread_mutex_unlock(mtx_); return item; } private: std::dequeT queue_; size_t capacity_; pthread_mutex_t mtx_; pthread_cond_t cond_; };单条件变量版本的逻辑是只要队列状态发生变化从空变成非空、从满变成非满就发一个 signal。所有线程醒来后拿锁、检查 while 条件条件不满足就继续睡满足就执行。这样写优势是简单你不需要区分“空”和“满”两种事件一个 cond 全部通知。2.3 单条件变量方案的两个隐藏问题第一个问题是惊群效应。如果队列里同时有多个生产者和多个消费者pthread_cond_signal只唤醒一个线程这本来是好事。但单条件变量的问题在于你没法精确决定唤醒的是哪一方。举个例子队列满了所有生产者都在等此时消费者取走一个元素调用 signal系统从等待队列里挑一个线程唤醒——它有可能挑中另一个消费者。这个消费者醒来发现队列还是空的只好继续睡。而真正该被唤醒的生产者还在等。信号被浪费了如果极端情况下每次 signal 都唤醒错人系统可能陷入“生产者一直等不到通知”的假死状态。用 broadcast 全换行不行更糟所有线程都醒来抢锁锁竞争激烈绝大多数人抢到锁后条件不满足又睡回去白白引发一轮 CPU 消耗。第二个问题其实是设计层面的单条件变量把“满”和“空”两种等待混在一起。从语义上讲队列空时应该唤醒生产者去生产队列满时应该唤醒消费者去消费这两组人关注的是不同方向的状态变化。混用一个 cond你就很难做定向唤醒只能拿吞吐量换正确性。所以在生产级别的实现里我更推荐用两个条件变量分别表达“非空”和“非满”也就是下面要讲的环形队列模型。3. 环形队列的 CP 模型两个条件变量各司其职3.1 环形队列和阻塞队列比优势在哪标准库的std::deque作为任务队列确实省心但它在高频生产消费场景下有个致命伤队列里的元素会被反复搬运、分配和释放内存碎片和 cache miss 会随着队列长度增加越来越明显。做高性能队列时我一般优先考虑环形缓冲区。环形队列本质是一个固定大小的数组维护头尾两个下标。元素入队时写在尾部出队时从头部取。当尾部下标追上头部时就是“满”头部追上尾部就是“空”。因为是固定数组不需要动态分配内存入队出队都是 O(1)对 cache 也很友好。3.2 两个条件变量的分工设计队列只有两种边界状态需要关注空消费者不能取必须等生产者满生产者不能放必须等消费者。所以一个条件变量管“非空”not_empty用于消费者等待另一个管“非满”not_full用于生产者等待。消费者取完元素后只需要 signal not_full让可能阻塞的生产者醒来生产者放入元素后只需要 signal not_empty让可能阻塞的消费者醒来。谁也别乱叫醒谁惊群概率大大降低。这背后的原则值得多说一句条件变量的数量应该等于需要区分的等待条件数。你等待的是“非空”还是“非满”在语义上是完全不同的两件事合用一个变量属于偷懒。单条件变量不是不能用但你想保证系统在复杂并发下依然高效就必须拆开。3.3 基于环形队列的完整实现我写一个环形队列版的生产者消费者模型整体结构尽量简洁方便你直接改造成自己的组件#include pthread.h #include vector templatetypename T class RingQueue { public: RingQueue(size_t cap) : capacity_(cap), data_(cap), head_(0), tail_(0), count_(0) { pthread_mutex_init(mtx_, nullptr); pthread_cond_init(not_empty_, nullptr); pthread_cond_init(not_full_, nullptr); } void push(const T item) { pthread_mutex_lock(mtx_); while (count_ capacity_) { pthread_cond_wait(not_full_, mtx_); // 只等“非满” } data_[tail_] item; tail_ (tail_ 1) % capacity_; count_; pthread_cond_signal(not_empty_); // 只通知“非空” pthread_mutex_unlock(mtx_); } T pop() { pthread_mutex_lock(mtx_); while (count_ 0) { pthread_cond_wait(not_empty_, mtx_); // 只等“非空” } T item data_[head_]; head_ (head_ 1) % capacity_; --count_; pthread_cond_signal(not_full_); // 只通知“非满” pthread_mutex_unlock(mtx_); return item; } private: std::vectorT data_; size_t capacity_; size_t head_; size_t tail_; size_t count_; pthread_mutex_t mtx_; pthread_cond_t not_empty_; pthread_cond_t not_full_; };这个实现只有几点需要留意head_、tail_、count_这三个变量都在锁内被修改外部无锁读取的行为是未定义的不要学网上某些写法用原子变量替代 count_ 再做无锁判空——在你没有足够内存屏障功底之前这是给自己挖坑。环形数组在写data_[tail_] item时如果 T 是可拷贝构造的注意异常安全性。生产级代码应该让 T 支持移动语义或者存储std::unique_ptrT这类资源句柄。两个条件变量 定向 signal 之后生产者不需要唤醒生产者消费者不需要唤醒消费者线程无效唤醒的量会明显下降。3.4 实测时的观察要点把队列容量设成 1你会看到生产者和消费者严格交替执行这时能直观验证两个条件变量是否正确协作。再把容量调大比如 1024放 4 个生产者、4 个消费者同时跑观察吞吐量。我自己的习惯是写一个带时间戳的压测脚本统计每秒钟成功 push 和 pop 的次数。对比单条件变量版本两个条件变量版本在 8 线程以上时吞吐提升通常在 20% 到 40% 之间线程数越多差距越明显。如果你的测试结果相反优先检查是不是你的临界区里还做了别的耗时操作比如打印日志或者 push 时不小心 copy 了大对象。4. 线程池把条件变量、互斥锁、任务队列组装成生产级组件4.1 为什么需要线程池每次需要处理任务就pthread_create一个线程用完再pthread_join销毁这在任务量小的时候没问题但任务一来就是几千个的时候线程创建销毁的开销会占到总耗时的相当比例。线程创建涉及内核分配栈空间、建立调度实体销毁还要回收这些都是真金白银的成本。线程池的思路是预先创建好一组线程不销毁不断从任务队列里取任务执行。线程本身是复用的任务队列负责流量缓冲。线程池三件套就是任务队列、互斥锁、条件变量外加一个工作线程数组。这正好把我们前面讲的所有同步原语串起来了。4.2 一个可用的线程池骨架设计线程池时我会把“任务执行”放到锁外。为什么任务本身可能耗时很长如果放在锁内执行其他线程想投递任务会被锁挡住整个池子的并发能力会崩塌。更严重的是任务内部如果再尝试往同一个池子投递任务可能出现死锁持有锁的任务在等锁另一个持有锁的投递线程又在等第一个任务释放锁互相卡死。骨架大概是这样的结构#include pthread.h #include queue #include vector #include functional class ThreadPool { public: ThreadPool(int threads) : shutdown_(false) { pthread_mutex_init(mtx_, nullptr); pthread_cond_init(cond_, nullptr); workers_.reserve(threads); for (int i 0; i threads; i) { pthread_t tid; pthread_create(tid, nullptr, worker_func, this); workers_.push_back(tid); } } void submit(std::functionvoid() task) { pthread_mutex_lock(mtx_); tasks_.push(std::move(task)); pthread_cond_signal(cond_); pthread_mutex_unlock(mtx_); } void shutdown() { pthread_mutex_lock(mtx_); shutdown_ true; pthread_cond_broadcast(cond_); pthread_mutex_unlock(mtx_); for (auto tid : workers_) { pthread_join(tid, nullptr); } } private: static void* worker_func(void* arg) { ThreadPool* pool static_castThreadPool*(arg); pool-work_loop(); return nullptr; } void work_loop() { while (true) { std::functionvoid() task; { pthread_mutex_lock(mtx_); while (tasks_.empty() !shutdown_) { pthread_cond_wait(cond_, mtx_); } if (shutdown_ tasks_.empty()) { pthread_mutex_unlock(mtx_); return; } task std::move(tasks_.front()); tasks_.pop(); pthread_mutex_unlock(mtx_); } task(); // 关键任务必须在锁外执行 } } std::queuestd::functionvoid() tasks_; std::vectorpthread_t workers_; pthread_mutex_t mtx_; pthread_cond_t cond_; bool shutdown_; };工作线程的执行逻辑可以用一句话概括没有任务就睡有任务就抢拿到任务先释放锁再执行。任务投递方的逻辑更简单加锁、塞队列、发通知、解锁。4.3 销毁线程池的规范流程线程池销毁是很多人容易翻车的地方。直接pthread_cancel工作线程我劝你别这么干容易造成任务执行一半被中断资源没来得及释放。规范的销毁流程是设置shutdown_ truepthread_cond_broadcast唤醒所有阻塞在条件变量上的工作线程逐一pthread_join回收线程保证队列里残留的任务要么被丢弃要么由某个线程执行完再退出。工作线程退出前必须重新加锁再检查shutdown_ tasks_.empty()这是因为可能会有“你刚把任务塞进队列还没 signalshutdown 就把大家全叫醒”的情况。如果 workers 醒来发现队列里有任务哪怕 shutdown 已经置位也应该先把任务执行完再退否则任务会静默丢失。4.4 线程数到底设多少合适这是一个老生常谈的问题。CPU 密集型任务线程数一般设为核数 1IO 密集型任务因为线程经常阻塞在 IO 上可以多一些常见经验值是核数 * 2左右。但真实系统里任务往往混合了 CPU 和 IO所以我更建议直接做一次压力测试先设 4、8、16、32 跑一轮观察吞吐量和响应延迟找到拐点。线程池的大小还会影响条件变量的竞争程度。池子过大任务队列很短时线程们就会频繁空转抢锁。我曾经在一个网络服务里把线程池从 32 改成 16吞吐反而高了 15%就是因为锁竞争减少了。所以“池子越大越好”这个直觉在并发场景下经常是错的。5. 线程安全锁只是手段不是银弹5.1 线程安全到底在说什么很多人把“线程安全”等同于“加锁”其实远远不止。线程安全指一段代码在多线程并发调用下仍然能维持正确的状态和可预期的结果。它的三个基础维度是原子性、可见性、有序性。原子性说的是一个操作不可被线程调度打断。比如count在 CPU 指令层面就不是原子的多个线程同时执行它就会互相覆盖。互斥锁能保证临界区的原子性但原子性不一定必须靠锁原子变量也可以。可见性说的是一个线程修改了数据另一个线程能不能立刻看到。CPU 有缓存编译器有优化你在这边改了变量另一个核上的线程可能读到的还是旧值。这就是为什么锁释放后其他线程能立刻看到最新状态——锁的实现自动带了内存屏障。有序性更隐蔽为了性能编译器和 CPU 都可能重排指令。单线程下没问题多线程下就可能产生意外顺序。所以为什么“不加锁的共享变量”很难写对就是因为这三个维度都难以保证。5.2 “不加锁”也能线程安全的情况在动手加锁之前先问自己这个数据真的需要锁吗如果它是只读的比如初始化后不变的路由表完全可以不加锁所有线程并发读就行。如果它只在某个线程内部使用那就做成局部变量或者搞一个线程局部存储根本不共享也谈不上安全。我再强调一种值得刻意使用的思路所有权转移。生产者创建了一个任务对象把它塞进队列消费者从队列里取出来它的“所有权”用完后销毁。在整个生命周期里一个对象同时只有一个线程持有。这样你不需要给对象本身加锁只需要保护队列的入队和出队操作。这是非常多高性能服务里常用的做法。5.3 volatile 背不了锁的锅我见过不少新手拿volatile当线程安全工具用把共享变量都标成 volatile以为就万事大吉了。volatile 确实能告诉编译器不要优化掉这个变量的读写但它不能保证 CPU 缓存之间的可见性也不能保证操作的原子性。两个线程同时对 volatile 变量做照样丢更新。真的需要无锁读写的场景请用 C11 的std::atomic或者 Linux 的__atomic内建函数。至少要知道只有原子变量的读写才是安全的普通变量即使加了 volatile 也不行。5.4 排查线程安全问题的经验线程安全问题的特点是难复现。数据竞争往往在特定调度顺序下才触发你调试时跑得好好的上线后偶发崩溃。我的排查思路是这样的先开线程消毒器ThreadSanitizer它能检测数据竞争编译时加-fsanitizethread跑一轮压力测试通常会直接报出有竞争的行号。再看崩溃栈如果崩溃点在一个看起来“不可能空”的指针上九成是别的地方数据竞争把内存写坏了顺着共享变量往上查。最后复盘设计有多少变量是跨线程共享的这些变量是不是都有明确的锁保护有没有可能通过所有权转移或不可变设计消除共享想清楚这三点再加锁也来得及。6. 读写锁读多写少场景下的第一选择6.1 互斥锁把读和写一视同仁太浪费了假设你有一个配置表几十个线程都在读只有极少数时刻有一个线程去更新。如果全用互斥锁每个读者都要和其他读者互相排斥明明读操作不改数据却被逼着串行执行并发能力白白浪费了。Linux 提供的pthread_rwlock_t就是为了解决这类场景允许多个读者同时持有读锁但写锁是排他的写锁持有期间没有任何读者或写者能进入临界区。符合直觉的说法是读读共享、读写互斥、写写互斥。基本用法不复杂pthread_rwlock_t rwlock; // 读路径 pthread_rwlock_rdlock(rwlock); // 读取共享数据 pthread_rwlock_unlock(rwlock); // 写路径 pthread_rwlock_wrlock(rwlock); // 修改共享数据 pthread_rwlock_unlock(rwlock);6.2 读写锁也会踩坑写者饥饿与锁升级用读写锁之前必须知道一个致命缺陷不同实现的调度策略不一样。有的实现是读者优先只要不断有读者到来写者就可能一直等不到锁这叫写者饥饿。某些实现里线程无法保证在持有读锁的情况下安全升级为写锁因为如果两个读者同时想升级就会互相等对方释放读锁死锁。所以绝对不要在一个线程内先 rdlock 后直接 wrlock要么统一用写锁要么先解锁再重新请求写锁但要接受中间状态可能被其他读者看到。我在实际项目里遇到过更隐蔽的问题超过一定线程数后读写锁的性能甚至不如互斥锁。原因是读写锁内部需要维护读者计数这本身就有原子操作和缓存一致性开销。如果临界区很短读者频繁加锁解锁原子操作的开销就可能大于锁竞争的开销。所以我建议用读写锁前先做个性能对比别想当然。6.3 读多写少更优雅的替代方案Copy-On-Write在某些场景下连读写锁都不需要。比如一份配置数据读者很多写者很少我常用的方案是读写加引用写者先把旧数据复制一份在副本上修改然后用原子操作替换掉指向数据的指针读者通过原子加载拿到当前指针读取数据。只要读者拿到指针后不再等待新版本老版本在最后一个读者释放引用前不会被回收。这是典型的无锁读优化比读写锁更适合“读频率超高、写频率极低”的场景。这个方案有一个前提数据本身是可复制且比较小的复制成本你承受得起。如果是一个巨大的缓存哈希表每次写都复制整个表代价可能比读写锁还高。所以还是要分场景没有银弹。6.4 读写锁适用性自查清单我给自己总结了四个条件满足才考虑用读写锁读操作远多于写操作比例至少 10:1每次持有读锁的时间比加锁本身的开销长得多临界区的读操作真的不会修改数据写操作频率不高但发生时能接受一定程度地阻塞读者。如果这四条里有一条不满足老老实实用互斥锁或者干脆走无锁路线别让读写锁的理论优势骗了你。7. 常见问题速查表与避坑清单7.1 典型问题与排查方向我在各种项目里反复遇到过的问题整理成一张表你排查的时候可以对号入座。现象可能原因排查要点程序卡死不打印任何日志条件变量等待后没人唤醒检查 signal/broadcast 是否在锁内调用检查条件变量是否被多个条件共用偶发“信号丢失”导致线程永远等wait 前用的是 if 而不是 while改用 while 循环重查条件多消费者场景下明明有任务却没人处理signal 唤醒的是错误方一个条件变量管一个语义生产者只 signal 非空消费者只 signal 非满线程池销毁后还有线程卡住shutdown 标志位没置上销毁前先锁内置 shutdown_再 broadcast再 join加锁后性能反而下降临界区过长或锁粒度太大只保护共享数据的修改任务执行移出锁外数据没加锁却能稳定复现 bug编译器和 CPU 重排了指令用原子变量或锁提供内存屏障不要裸用 volatile7.2 条件变量使用的七个铁律最后分享我在实际调试中总结出来的一套条件变量使用纪律每次写代码都对照一遍条件变量的等待必须放在锁内条件判断必须用 while不能用 if只要修改了条件相关的共享状态就必须发送对应信号signal 和 broadcast 都要在锁内或锁外调用可以在锁外调用但必须在修改共享状态后调用一个条件变量对应一种等待语义别混用唤醒后必须重新检查条件再往下走程序退出前必须唤醒所有等待者否则线程会一直挂起。如果你能用好互斥锁解决“互斥”再用条件变量解决“等待”把阻塞队列和环形队列的 CP 模型吃透最后把线程池封装成顺手的基础组件Linux 多线程开发就算是真正入门了。至于读写锁和线程安全更多是设计层面的权衡——你知道得越多越明白锁不是越多越好而是越精准越好。我个人在实践里总把这句话挂在嘴边能用“所有权转移”解决就不加锁能用“无锁读”解决就不用读写锁必须加锁时再想条件变量怎么配。多线程程序的复杂度和你能意识到的并发问题数是成正比的把所有隐藏的共享状态都摆到桌面上来很多奇奇怪怪的崩溃都能在写代码阶段就避免掉。希望你读完这篇不只是记住了几个 API而是能在下一次写多线程模块时先把“谁在等、谁唤醒谁、条件怎么判断”这三件事想明白。这套思路用熟了比你背口诀管用得多。
阅读完成 · 觉得有帮助？
