我在最早写多线程程序的时候曾经特别想当然地以为「给共享变量加上互斥锁程序就安全了」。结果联调测试的时候数据确实不乱了但业务节奏全乱了某个线程等的数据明明已经被另一个线程准备好了它却还在傻乎乎地轮询一个标志位CPU 被白白吃满程序行为完全不可控。后来我才意识到线程之间除了「互相排斥」的关系还有大量「一个干完另一个才能开始」的协作关系。这也就是标题里说的线程同步。这一篇是《线程同步与互斥》的第二部分会围绕几个核心主题展开条件变量、基于阻塞队列的生产者消费者模型、基于环形队列的生产者消费者模型、线程池、线程安全与读写锁。它适合正在补 Linux 面经基础的人也适合真正要在 C/C 工程里写多线程代码、却总觉得「锁加上了但行为不对」的开发者。这一篇的目标很朴素让你搞懂「互斥」之外的另一半 —— 等待、唤醒、协作以及这些机制在真实模型里是怎么配合的。1. 为什么有了互斥锁还得专门讲线程同步先聊一个最本质的问题。很多人学完互斥锁之后会有一种错觉我只要在操作共享数据前后加上 lock 和 unlock多线程就安全了。这句话对一半因为互斥锁解决的是「同时访问」的问题它不解决「先后顺序」的问题。举一个很常见的例子有两个线程一个是生产者负责往缓冲区里写数据一个是消费者负责从缓冲区里取数据。消费者的逻辑很自然如果缓冲区是空的我就等待等缓冲区有数据了我再去取。在没有同步原语的情况下你只能写成轮询while (buffer_is_empty()) { // 什么也不做继续转圈 }这段代码从逻辑上讲没错但它非常糟糕。每次循环都要去访问共享变量生产者持锁写入的时候消费者在空转抢锁CPU 占用直接拉满更麻烦的是你很难确定轮询的间隔到底设多少间隔长了数据延迟高间隔短了 CPU 损耗大。本质上轮询靠的是「反复检查」而同步机制靠的是「事件通知」。操作系统提供了条件变量、信号量这些原语就是为了让线程能在条件不满足时真正睡下去而不是在条件不满足时死磕。再看一个更隐蔽的问题互斥锁保证的是「临界区互斥进入」它只管理访问不管理依赖。比如线程 A 要先算出一个结果线程 B 才能拿着这个结果继续执行。如果只用互斥锁你是没办法优雅表达「B 必须等 A 完成」这种依赖关系的。你可以在 B 里循环判断结果标志位但这就又回到了轮询的泥潭。所以线程同步要解决的本质上是事件次序问题也就是让线程在某些条件不满足时阻塞在条件满足时被唤醒并且不会因为检查条件和等待状态之间存在时间差而丢失唤醒信号。这个「时间差」非常关键也是后面条件变量设计中核心要考虑的点。理解了这个背景后面的所有模型就都好理解了。2. 条件变量让线程学会「等消息再干活」条件变量condition variable是 Linux 下最常用的线程同步原语之一。它本身不承担互斥职责它做的事情只有两件让线程等一个条件以及让线程在条件可能满足时唤醒等待者。2.1 几个核心 API 和一个经典陷阱先看 API函数作用重要说明pthread_cond_init初始化条件变量也可以用PTHREAD_COND_INITIALIZER静态初始化pthread_cond_wait等待条件满足必须传入一个已加锁的互斥锁pthread_cond_signal唤醒一个等待者如果没人等本次唤醒作废pthread_cond_broadcast唤醒所有等待者一般配合「条件变化影响多个线程」的场景pthread_cond_destroy销毁确认无线程使用后再销毁这里最劝退新人的一点是pthread_cond_wait为什么非要配一个互斥锁因为条件变量的使用场景里必然涉及「检查条件」和「进入等待」这两个动作。如果这两个动作之间没有互斥保护就会发生经典竞态消费者线程检查条件发现「没有数据」正准备进入等待就在这一瞬间生产者线程放入数据并发送了 signal。由于消费者还没有真正睡下去signal 没人接收于是消费者继续等待而数据已经错过了 —— 这个唤醒信号永远丢了。pthread_cond_wait的另一个隐藏行为是它内部会自动释放传入的互斥锁让出 CPU被唤醒之后它会先重新获取互斥锁然后才从函数返回。所以你会发现一个有趣的事实从调用者的角度看pthread_cond_wait前后的状态都是「持锁状态」只有它内部沉睡的间隙锁是被释放的。2.2 为什么 wait 外面必须套 while 而不是 if这是被问烂了但还是有无数人写错的点。教科书里生产者消费者的等待逻辑标准写法是pthread_mutex_lock(mtx); while (!data_ready) { pthread_cond_wait(cond, mtx); } // 这里可以安全地消费数据 pthread_mutex_unlock(mtx);为什么不能写成if (!data_ready)官方文档明确说条件变量可能存在「虚假唤醒」spurious wakeup。什么意思呢就是线程明明没有被 signal 或 broadcast却自己从 wait 中返回了。这主要是因为 Linux 的 futex 实现和信号干扰也可能是因为多线程同时等待时被唤醒的线程拿到锁之后发现条件已经被别的线程消费掉了。如果你用的是 if一旦条件过期你会拿着一个错误的前提继续执行程序行为完全不可控。而 while 的作用是醒来之后立刻重新检查条件如果条件还满足不了继续睡回去。这是一种防御性写法它在工程里是必须的因为多线程环境下「被唤醒」不代表「条件一定成立」只能代表「条件可能成立」。顺便说一句pthread_cond_broadcast的经典使用场景多个消费者线程在等同一个条件而条件一旦满足可能要唤醒所有线程重新竞争。比如「批量任务下达」你希望所有工人线程都起来干活而不是只叫一个。2.3 一个最小可跑的示范signal 唤醒顺序的验证我自己最早跑条件变量时最喜欢干的一件事就是打印线程的调度顺序。做个简单验证一个消费者等待三个生产者轮流唤醒。#include pthread.h #include stdio.h #include unistd.h pthread_mutex_t mtx PTHREAD_MUTEX_INITIALIZER; pthread_cond_t cond PTHREAD_COND_INITIALIZER; int data_ready 0; void *producer(void *arg) { long id (long)arg; sleep(1); pthread_mutex_lock(mtx); data_ready 1; printf(producer %ld set data_ready\n, id); pthread_cond_signal(cond); pthread_mutex_unlock(mtx); return NULL; } void *consumer(void *arg) { pthread_mutex_lock(mtx); while (!data_ready) { pthread_cond_wait(cond, mtx); } printf(consumer wake up, data_ready%d\n, data_ready); pthread_mutex_unlock(mtx); return NULL; } int main() { pthread_t c, p[3]; pthread_create(c, NULL, consumer, NULL); for (long i 0; i 3; i) { pthread_create(p[i], NULL, producer, (void *)i); } pthread_join(c, NULL); for (int i 0; i 3; i) { pthread_join(p[i], NULL); } return 0; }这里有一个细节值得注意signal 并不要求必须放在锁内调用但实践里我推荐放在锁内或者至少在释放锁之后再 signal 也行。真正要注意的是signal 的语义是「唤醒正在等待的线程」如果当前没有线程在等signal 就会空转一次什么也不会留下。这也就是为什么条件变量不能代替计数器你自己心里要清楚这个边界。3. 基于阻塞队列的 cp 模型最常见的生产者消费者写法条件变量单独用容易写散工程上更常见的做法是把它和队列封装在一起对外只暴露put和take两个接口。这就是阻塞队列。它是面试里出现频率最高、也是实用性最强的生产者消费者模型之一。3.1 为什么要封队列而不是裸写条件变量有一次我带了个实习生让他实现生产者消费者。他非常认真地直接写了三个线程在主函数里锁来锁去结果共享变量一多他自己也绕晕了。我后来跟他讲了一个做工程的原则多线程代码里共享状态越集中越容易保证正确性。阻塞队列就是这样一种「把共享状态收紧到一个类里」的设计。生产者和消费者不再直接面对条件变量它们只需要面对队列的两个方法。类内部维护容量、队列、两个条件变量所有线程安全细节对外不可见。一旦封装好你可以把它当积木去拼装更大的系统。3.2 封装一个自带两个条件变量的 BlockQueueC 版本可以写成这样#include condition_variable #include mutex #include queue #include iostream template typename T class BlockQueue { public: explicit BlockQueue(size_t cap) : cap_(cap) {} void put(const T data) { std::unique_lockstd::mutex lock(mtx_); // 队列满了生产者阻塞等待 while (q_.size() cap_) { not_full_.wait(lock); } q_.push(data); // 入队后消费者可能正在等数据 not_empty_.notify_one(); } T take() { std::unique_lockstd::mutex lock(mtx_); while (q_.empty()) { not_empty_.wait(lock); } T data q_.front(); q_.pop(); not_full_.notify_one(); return data; } private: std::queueT q_; size_t cap_; std::mutex mtx_; std::condition_variable not_full_; std::condition_variable not_empty_; };注意这里我用了两个条件变量not_full_和not_empty_。为什么不是一个因为生产者和消费者等的是不同的条件生产者等「队列有空位」消费者等「队列有数据」。如果用同一个条件变量每次唤醒都要让所有线程都检查一遍自己的条件对不对这也不会出错但会带来大量无意义唤醒。两个条件变量的成本很低收益却很清晰。3.3 让生产者和消费者跑起来主线程里可以这样组织BlockQueueint bq(5); void *producer(void *arg) { for (int i 0; i 20; i) { bq.put(i); std::cout put i std::endl; } return nullptr; } void *consumer(void *arg) { for (int i 0; i 20; i) { int v bq.take(); std::cout take v std::endl; } return nullptr; }跑完之后你会发现put和take的输出顺序是交错的甚至可能先出现后面的 take 再出现前面的 put但是全部 20 个数据都会被完整消费掉总数量守恒。这个「打印顺序乱但数量守恒」的现象其实就是多线程编程里最常见的感受不要试图控制全局时序只要保证共享数据的约束不被破坏。3.4 面试里经常追问的两个延伸点第一个延伸点如何让某个生产者优先于其他生产者或者某个消费者特别敏感答案是在put/take里增加优先级条件。比如给队列的元素包装成带优先级的结构体在 put 里用优先级队列替代普通队列或者在生产者 put 时对某个特殊标志位做检查。条件变量本身不负责调度策略它只负责阻塞和唤醒策略由业务逻辑决定。第二个延伸点阻塞队列的容量设置有什么讲究太小了容易频繁触发生产者阻塞吞吐量上不去太大了会积压任务实时性变差。实践中一个常见做法是先设一个经验值比如按任务吞吐量和平均处理时间的乘积估算也就是 Littles law队列长度 ≈ 平均到达速率 × 平均处理时间再压测调优。这个公式虽然不是万能的但作为初始值非常靠谱。4. 基于环形队列的 cp 模型用信号量管理「空位」和「数据」阻塞队列的思路是用条件变量等待「空位」和「数据」而环形队列配合信号量的模型是从另一个角度解决同一件事信号量天然就是一个计数器特别适合表示「剩余资源数」。4.1 为什么我会在第二个模型里换用信号量面试里经常让人对比阻塞队列模型和环形队列模型。我的理解是这样它们都能实现生产者和消费者的同步但思路不同。阻塞队列模型的核心是「条件等待」生产者和消费者都等待一个特定的布尔条件而环形队列 信号量模型的核心是「资源计数」把队列的空位看作一份资源把队列中的数据看作另一份资源。信号量有两个基本操作sem_wait负责把计数器减一如果计数器为 0线程阻塞sem_post负责把计数器加一并唤醒可能正在等待的线程。这里有个绝妙的对应关系sem_empty表示空位数量初始值等于环形队列容量sem_full表示数据数量初始值等于 0生产者每一次生产就是「申请一个空位」sem_wait(sem_empty)然后在空位上写入数据最后「释放一个数据」sem_post(sem_full)。消费者反过来先「申请一个数据」再「释放一个空位」。这是教科书里最优雅的对称结构理解之后很难忘掉。4.2 一个完整可用的环形队列实现这里要注意一个细节环形队列的写入位置和读取位置分别由生产者和消费者各自维护所以理论上「生产者写的位置」和「消费者读的位置」是两个不同的变量。虽然它们都在队列里但它们被访问的时机完全不同。所以工程上通常给生产者的下标加一把锁给消费者的下标加另一把锁两条访问路径各自互斥互不干扰。#include semaphore.h #include vector #include pthread.h template typename T class RingQueue { public: explicit RingQueue(int cap) : cap_(cap), vec_(cap) { sem_init(sem_empty_, 0, cap); sem_init(sem_full_, 0, 0); pthread_mutex_init(p_mtx_, nullptr); pthread_mutex_init(c_mtx_, nullptr); p_idx_ 0; c_idx_ 0; } void put(const T data) { sem_wait(sem_empty_); // 申请一个空位 pthread_mutex_lock(p_mtx_); vec_[p_idx_] data; p_idx_ (p_idx_ 1) % cap_; pthread_mutex_unlock(p_mtx_); sem_post(sem_full_); // 增加一个数据 } T take() { sem_wait(sem_full_); // 申请一个数据 pthread_mutex_lock(c_mtx_); T data vec_[c_idx_]; c_idx_ (c_idx_ 1) % cap_; pthread_mutex_unlock(c_mtx_); sem_post(sem_empty_); // 增加一个空位 return data; } private: size_t cap_; std::vectorT vec_; sem_t sem_empty_; sem_t sem_full_; pthread_mutex_t p_mtx_; pthread_mutex_t c_mtx_; int p_idx_; int c_idx_; };这里有一个初学者容易犯的错误给整个环形队列只加一把锁。这么做没错但会把生产者和消费者串行化丢失了「生产者写头部、消费者读尾部」本来可以并行推进的优势。双锁的精髓在于只要队列没有满到边界生产者和消费者访问的其实是不同的数组位置它们之间的竞争被解耦了。4.3 环形队列模型在工程里的真实应用很多人觉得环形队列只是面试题实际开发里没什么用这其实是个误解。凡是「生产快、消费慢」或者相反、且希望空间固定的场景环形队列几乎是标配。比如音频采集播放系统里的缓冲区采集线程往环形缓冲区写 PCM 数据播放线程从缓冲区读数据空间预先分配好运行期零动态内存分配。再比如网络库里的收包缓冲区、日志异步落盘前的内存缓冲很多都是环形队列的思路。它的一个核心优势是不需要动态申请和释放内存队列容量固定数据覆盖策略还可以灵活设计比如允许丢最旧的数据。配合信号量之后读写线程之间天然知道有多少数据可用不需要额外维护一个 size 变量 —— 因为信号量本身就是 size。5. 线程池把 cp 模型封装成生产级工作引擎如果说前面的阻塞队列和环形队列是「零件」那么线程池就是把「任务队列 一组工作线程 生命周期管理」整合起来的生产级引擎。它在后端服务、嵌入式 Linux 应用、并发测试工具里都极其常见。5.1 线程池到底解决了什么问题频繁创建和销毁线程的开销是很大的。一次pthread_create的背后涉及内核线程创建、栈分配、调度器接入如果任务本身很轻量比如只是算一个哈希值那可能「创建线程的代价 执行任务的代价」。线程池的思路很直接提前创建好一批线程它们空闲时就阻塞等待任务来了就取一个任务执行执行完继续等待线程本身不销毁。除了省创建开销线程池还有一个隐形优势它天然限制了并发度避免无节制地开线程把系统资源耗尽。这一点在嵌入式 Linux 场景里特别重要内存有限文件描述符有限线程数量一旦失控系统可能直接 OOM 或者调度崩溃。5.2 一个简单却完整的线程池骨架我习惯用 C11 的std::thread和std::condition_variable写线程池代码比 C 的 pthread 版本清晰很多关键逻辑依然依赖前两节讲的同步原理#include condition_variable #include functional #include iostream #include mutex #include queue #include thread #include vector class ThreadPool { public: explicit ThreadPool(size_t threads) { for (size_t i 0; i threads; i) { workers_.emplace_back([this] { while (true) { std::functionvoid() task; { std::unique_lockstd::mutex lock(mtx_); cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); }); if (stop_ tasks_.empty()) { return; } task std::move(tasks_.front()); tasks_.pop(); } task(); // 在锁外执行避免持锁运行业务代码 } }); } } template typename F void enqueue(F f) { { std::unique_lockstd::mutex lock(mtx_); tasks_.emplace(std::forwardF(f)); } cv_.notify_one(); } ~ThreadPool() { { std::unique_lockstd::mutex lock(mtx_); stop_ true; } cv_.notify_all(); for (auto worker : workers_) { worker.join(); } } private: std::vectorstd::thread workers_; std::queuestd::functionvoid() tasks_; std::mutex mtx_; std::condition_variable cv_; bool stop_ false; };使用方式非常直白ThreadPool pool(4); for (int i 0; i 10; i) { pool.enqueue([i] { std::cout task i running in thread std::this_thread::get_id() std::endl; }); } // 析构时会等待所有任务执行完注意std::functionvoid()能装下任意可调用对象所以 lambda、函数指针、甚至 bind 表达式都能提交进来。这个设计虽然简单但已经覆盖了线程池 90% 的核心逻辑。5.3 线程池配置里最常被问的三个参数问题第一个核心线程数设多少没有万能公式但有两个经验方向。如果是 CPU 密集型任务建议设置成std::thread::hardware_concurrency()也就是物理核心数左右避免线程切换浪费算力。如果是 IO 密集型任务线程在等 IO 时 CPU 是空闲的可以把线程数调高一些常见经验是 CPU 核数的两倍甚至更多。总的来说按「CPU 密集看核数IO 密集看等待比」这个方向先给初始值再做压测。第二个任务队列用无界队列还是有界队列无界队列实现简单但风险是任务积压到内存爆炸有界队列配合「饱和策略」更可控。所谓饱和策略就是队列满时怎么办直接丢弃、丢弃最旧任务、调用者负责执行、或者阻塞调用者。你要是去面试能主动说出这个「饱和策略」的概念通常能加分不少。第三个提交任务和唤醒线程的顺序谁的坑最多我最常看到的问题是把notify_one写在了持锁状态下然后才发现某个任务处理很慢其他任务全部饿死。notify_one放在锁内并不是错误但如果你对唤醒时机比较敏感我建议把任务入队之后立刻释放锁再通知线程池里的线程去取任务。这样能让唤醒动作尽早发生减少「线程已经从等待中被唤醒了却还要等锁」的时间。5.4 线程池的几个防崩细节线程池写好之后我建议重点检查这三个场景。场景一线程执行的任务抛异常。std::thread里如果任务抛异常且未被捕获会直接调用std::terminate终结整个进程。所以业务代码建议在自己的任务函数里 try-catch或者在线程池内部包裹一层兜底逻辑。场景二析构时stop_的可见性。析构函数里要先置stop_ true再notify_all。顺序反了可能导致部分线程已经睡下去永远等不到唤醒析构时join卡死。场景三唤醒丢失。enqueue时先加锁再入队出锁后再notify_one这是最不容易出错的顺序。如果你把 notify 放在入队之前那线程被唤醒后可能发现队列还是空的又睡回去结果任务已经入队了却没人通知 —— 经典唤醒丢失。6. 读写锁读多写少场景下的优化之道前面讨论的锁都是「互斥」语义任何时刻只有一个线程能持有锁。但在很多真实系统里读操作远比写操作频繁比如配置中心的路由表、DNS 缓存、嵌入式设备里的传感器校准参数。如果对读操作也互斥那多个读者明明可以并行读取却被强行串行化白白损失性能。6.1 读写锁的核心原理读写锁rwlock放宽了互斥约束多个读者可以同时持有锁但写者必须独占。也就是说「读者和读者之间共享写者和任何其他线程互斥」。这个语义刚好匹配「读多写少」的场景pthread_rwlock_t rwlock PTHREAD_RWLOCK_INITIALIZER; void read_config() { pthread_rwlock_rdlock(rwlock); // 多个读者可以同时进入这里 pthread_rwlock_unlock(rwlock); } void update_config() { pthread_rwlock_wrlock(rwlock); // 写者进入时既不允许其他写者也不允许读者 pthread_rwlock_unlock(rwlock); }用的时候要注意rdlock之间是共享的所以你不应该在持有读锁时去修改任何共享数据wrlock是独占的但它和普通互斥锁依然有区别 —— 它允许「当一个写者在队列里等待时后续的读者要排队让行」具体行为取决于实现策略。6.2 读者优先和写者优先的取舍Linux 下 glibc 默认的pthread_rwlock是写者优先writer preference的也就是只要有写者在等待新的读者会被挡住防止写者饿死。这个策略很合理因为写者一旦长期得不到锁数据永远更新不了读者的读也失去意义。但你在阅读某些教学资料或者面试被问时会碰到讨论「读者优先」的场景读者一直来写者一直等不到锁。这种策略会让读操作永远保持低延迟代价是写操作可能被无限期推迟。工程上一旦发现某个锁出现「写者饥饿」就应该考虑改成写者优先或者直接使用 C14 的std::shared_timed_mutex、C17 的std::shared_mutex它们同样支持共享锁和独占锁的语义而且用法更现代#include shared_mutex std::shared_mutex rw_lock; void read_config() { std::shared_lockstd::shared_mutex lock(rw_lock); // 多个读者共享 } void update_config() { std::unique_lockstd::shared_mutex lock(rw_lock); // 写者独占 }6.3 读写锁真的一定比互斥锁快吗不一定。读写锁的临界区如果非常小比如只读一个 int那么读写锁内部的读者计数、写者优先排队逻辑造成的开销可能比普通互斥锁还大。你拿到的「性能优势」会被复杂的锁内部状态抵消。所以我自己的经验是只有在临界区里确实有足够的读操作且读的比例显著高于写比如 5:1 以上时才值得上读写锁。否则普通互斥锁更简单、更可控。另外还有一个被讨论很多的选择题自旋锁还是互斥锁还是读写锁我的建议是应用层代码优先用互斥锁和读写锁不到万不得已别碰自旋锁。自旋锁的适用场景是「临界区极短 持锁时间远小于线程切换时间」比如内核态里操作一个链表节点。在用户态你很难精准控制临界区时长睡眠唤醒型锁通常更稳妥。写在最后的一点体会把条件变量、阻塞队列、环形队列、线程池和读写锁串起来看你会发现它们的共同点都在于管理好「共享数据的访问次序」而不是急着去「让所有线程一起跑」。真正的工程代码里很少有一个线程从头到尾独占资源更多时候是一群线程围绕几个共享结构体做协作。你能把「谁在等待什么条件」「谁在唤醒谁」「锁的粒度到底该有多大」这三件事想清楚多线程程序就不太会出大毛病。我个人的建议是碰到多线程问题先画一下共享数据流线程 A 往哪里写线程 B 从哪里读哪个变量是它们之间的边界。然后再决定用条件变量、信号量还是读写锁。工具本身不难难的是搞明白自己到底在等哪个条件以及这个条件的检查必须和等待动作连成一个整体。这是我在踩过很多次坑之后最深的一点体会。