上一篇聊线程互斥时我收到最多的留言是这类的互斥锁我理解了但锁被占用的时候另一个线程凭什么知道自己该等着而不是反复来敲门锁一解开等待的线程又是怎么被叫醒的这些追问其实指向同一个话题——线程同步。所以这篇我专门把同步这条线讲透标题里的“cp模型”不是啥玄学就是 consumer-producer生产者消费者模型条件变量、基于阻塞队列和环形队列的两种 cp 模型、线程池、线程安全、读写锁全部串起来讲。看完你至少能自己手写一个线程池并且知道什么时候该用哪种同步原语而不是拿着互斥锁梭哈一切。内容会有点长但每段都是实际调代码时踩过的坑不是教科书复读。我按这条主线走先搞懂条件变量的等待/唤醒机制再搭阻塞队列版本的生产者消费者接着换信号量实现环形队列版本然后封装线程池最后补线程安全和读写锁的边界问题。每一步都有能直接编译运行的代码我会把参数和步骤背后的原因一并写清楚。1. 从互斥到同步条件变量把“锁”变成了“通知”1.1 一个轮询问题逼出了条件变量假设一个线程要从共享队列里取数据队列为空怎么办最笨的办法是加锁后循环检查while (queue_empty(q)) { pthread_mutex_unlock(lock); usleep(1000); // 睡眠后再试 pthread_mutex_lock(lock); }这段代码能工作但问题很大。usleep 的间隔不好选短了CPU 空转得厉害长了数据到了却有额外延迟。一个更好的办法是让消费者明确“睡觉”等生产者“打电话叫醒”于是条件变量登场。条件变量的核心是三个 APIpthread_cond_wait让线程阻塞等待某个条件成立pthread_cond_signal唤醒一个等待者pthread_cond_broadcast唤醒全部等待者。它必须和一个互斥锁配合使用原因很微妙判断“队列是否为空”这个动作本身需要锁保护而等待动作必须一次性完成“释放锁 进入睡眠”否则中间会出现竞态。1.2 wait 的原子性是理解条件变量的钥匙pthread_cond_wait(cond, mutex)做的事情用伪代码看是这样的// 伪代码wait 的内部逻辑 pthread_mutex_unlock(mutex); // 释放锁 block_on(cond); // 挂起线程等待被唤醒 pthread_mutex_lock(mutex); // 被唤醒后重新拿锁关键在于第三步等线程被唤醒时它并不知道当前条件是否真的满足。所以标准用法是“while 循环 条件判断”而不是“if 判断”。如果只判断一次万一发生了虚假唤醒spurious wakeup线程就可能拿到空数据或者越界访问。注意阻塞队列实现里生产者和消费者分别需要两个条件变量一个表示“队列不满”一个表示“队列非空”。千万不要用一个条件变量省事否则想唤醒消费者时可能误唤醒生产者虽然程序不死但性能会退化语义也不清晰。1.3 条件变量的标准使用框架不管是生产者还是消费者代码骨架都一样三步走加锁while (条件不满足) pthread_cond_wait(...)操作共享数据解锁生产者的唤醒条件pthread_cond_signal(not_empty)消费者空了队列应该pthread_cond_signal(not_full)。注意 signal 的时机必须在锁内调用吗标准答案是“可以不在锁内”但为了简单可靠我习惯在pthread_mutex_unlock之后再 signal。两种方式各有拥趸实际操作中差别不大关键是别在 wait 返回后忘记重新检查条件。2. 基于阻塞队列的 cp 模型最直观的生产者消费者2.1 阻塞队列的完整实现下面这个队列用互斥锁 两个条件变量实现支持一或多个生产者和消费者。我用 C 语言写方便对照 POSIX API#include pthread.h #include stdlib.h #include string.h typedef struct block_queue { int *buf; size_t capacity; size_t head, tail, count; pthread_mutex_t lock; pthread_cond_t not_full; // 生产者等待 pthread_cond_t not_empty; // 消费者等待 } block_queue; void bq_init(block_queue *q, size_t cap) { q-buf malloc(sizeof(int) * cap); q-capacity cap; q-head q-tail q-count 0; pthread_mutex_init(q-lock, NULL); pthread_cond_init(q-not_full, NULL); pthread_cond_init(q-not_empty, NULL); } void bq_push(block_queue *q, int val) { pthread_mutex_lock(q-lock); while (q-count q-capacity) { pthread_cond_wait(q-not_full, q-lock); } q-buf[q-tail] val; q-tail (q-tail 1) % q-capacity; q-count; pthread_cond_signal(q-not_empty); pthread_mutex_unlock(q-lock); } int bq_pop(block_queue *q) { pthread_mutex_lock(q-lock); while (q-count 0) { pthread_cond_wait(q-not_empty, q-lock); } int val q-buf[q-head]; q-head (q-head 1) % q-capacity; q-count--; pthread_cond_signal(q-not_full); pthread_mutex_unlock(q-lock); return val; }想要多生产多消费这个队列直接就能用因为所有操作都受同一把锁保护。但要注意bq_push里的 signal 并没有精确指定唤醒谁如果同时有多个消费者等待唤醒哪个由调度器决定这是符合预期的——队列本来就不该绑定特定消费者。2.2 条件变量的“丢失唤醒”陷阱网上很多简化版代码喜欢把while写成if单生产者单消费者场景下测试没问题一旦多线程竞争就可能在wait返回后另一个线程已经把唯一数据取走了导致读取越界或读到脏数据。另一种常见的丢失唤醒场景pthread_cond_signal调用时恰好没有线程在等待信号直接丢弃。这没问题因为信号本身的语义就是“此刻队列状态变了如果有人在等就叫醒他”。真正危险的反而是某些人试图加一个“等待中计数”来优化结果计数和加锁顺序没配对造成死锁。2.3 阻塞队列适合什么场景阻塞队列最适合“任务边界清晰、数据量波动大”的场景。比如 Web 服务器把 HTTP 请求塞进队列worker 线程从队列里取任务处理。队列天然起到了“削峰”的作用请求短时间爆发时生产者不会被压垮消费者慢慢消化。缺点是每次 push/pop 都要加锁高吞吐时会成为瓶颈。这时就该考虑下面这种基于原子索引和信号量的环形队列或者进一步降低锁粒度。3. 基于环形队列的 cp 模型用信号量表示资源数量3.1 信号量的思路完全不同环形队列和阻塞队列的核心差别是它用两个信号量分别记录“剩余可写空间数”和“可读数据数”而不是靠条件变量判断队列状态。信号量本身自带一个计数器sem_wait会原子地把计数器减一计数器为 0 时就阻塞sem_post原子地加一并唤醒一个阻塞者。初始化时empty信号量初始化为队列容量 Nfull信号量初始化为 0。生产者做事前先sem_wait(empty)得到一个空位消费者做事前先sem_wait(full)拿到一份数据。事情做完后再sem_post另一个信号量。3.2 完整实现单生产单消费版本#include semaphore.h #include pthread.h #include stdlib.h typedef struct ring_queue { int *buf; size_t capacity; size_t read_pos, write_pos; sem_t empty, full; } ring_queue; void rq_init(ring_queue *q, size_t cap) { q-buf malloc(sizeof(int) * cap); q-capacity cap; q-read_pos q-write_pos 0; sem_init(q-empty, 0, cap); sem_init(q-full, 0, 0); } void rq_push(ring_queue *q, int val) { sem_wait(q-empty); q-buf[q-write_pos] val; q-write_pos (q-write_pos 1) % q-capacity; sem_post(q-full); } int rq_pop(ring_queue *q) { sem_wait(q-full); int val q-buf[q-read_pos]; q-read_pos (q-read_pos 1) % q-capacity; sem_post(q-empty); return val; }单生产单消费场景下这个实现完全不需要互斥锁。原因是两个线程操作的分别是 write_pos 和 read_pos一个只在“空位被填满”之后才让消费者读另一个只在“数据被消费”之后才让生产者写信号量天然保证了顺序。这比条件变量版本少了锁竞争性能上一个量级。3.3 多生产多消费必须加锁保护索引如果生产者和消费者都不止一个两个线程同时执行q-write_pos (q-write_pos 1) % q-capacity就会有问题索引更新不是原子操作。解决办法是给索引更新加一个轻量锁void rq_push_mt(ring_queue *q, int val) { sem_wait(q-empty); pthread_mutex_lock(q-lock); q-buf[q-write_pos] val; q-write_pos (q-write_pos 1) % q-capacity; pthread_mutex_unlock(q-lock); sem_post(q-full); }这里的锁只保护索引更新不保护整个操作所以锁的持有时间极短。相比阻塞队列每次操作都要锁整个队列的 count、head、tail并发度明显更高这是能用环形队列尽量用环形队列的原因。心得写环形队列时最容易犯的错是忘记把write_pos也纳入锁保护只给read_pos加锁。结果是数据被覆盖debug 时特别难查因为并不是每次跑都出问题只有两个生产者恰好同时推进索引时才丢数据。建议先在多线程压力下反复跑配合 TSan 或-fsanitizethread验证。4. 线程池cp 模型的工程化封装4.1 为什么需要线程池一次线程创建的开销远比你想象的贵。线程的创建要经历内核分配 task_struct、建立栈空间、调度器入场等步骤。如果业务逻辑只跑 0.1 毫秒而线程创建销毁耗时 0.5 毫秒那还不如串行执行。线程池的思路是提前创建一批线程把任务丢进任务队列让这些线程反复取任务执行。支付线程创建开销一次之后全都是纯业务时间。线程池本质上就是一个“生产者消费者模型”外部提交任务的线程是生产者线程池里的工作线程是消费者任务队列是中间缓冲区。这个认知是写线程池的第一性原理。4.2 一个最小但完整的线程池实现以下实现固定线程数 N 个任务用函数指针加void*参数表示#include pthread.h #include stdlib.h typedef struct task { void (*func)(void*); void *arg; } task; typedef struct thread_pool { task *tasks; size_t queue_capacity; size_t head, tail, count; pthread_t *threads; size_t thread_count; pthread_mutex_t lock; pthread_cond_t not_empty; pthread_cond_t not_full; int shutdown; } thread_pool; void *worker_main(void *arg) { thread_pool *pool (thread_pool*)arg; while (1) { pthread_mutex_lock(pool-lock); while (pool-count 0 !pool-shutdown) { pthread_cond_wait(pool-not_empty, pool-lock); } if (pool-shutdown pool-count 0) { pthread_mutex_unlock(pool-lock); break; } task t pool-tasks[pool-head]; pool-head (pool-head 1) % pool-queue_capacity; pool-count--; pthread_cond_signal(pool-not_full); pthread_mutex_unlock(pool-lock); t.func(t.arg); // 在锁外执行任务 } return NULL; } void pool_init(thread_pool *pool, size_t threads, size_t qcap) { pool-tasks malloc(sizeof(task) * qcap); pool-queue_capacity qcap; pool-head pool-tail pool-count 0; pool-thread_count threads; pool-shutdown 0; pthread_mutex_init(pool-lock, NULL); pthread_cond_init(pool-not_empty, NULL); pthread_cond_init(pool-not_full, NULL); pool-threads malloc(sizeof(pthread_t) * threads); for (size_t i 0; i threads; i) { pthread_create(pool-threads[i], NULL, worker_main, pool); } } void pool_submit(thread_pool *pool, void (*func)(void*), void *arg) { pthread_mutex_lock(pool-lock); while (pool-count pool-queue_capacity !pool-shutdown) { pthread_cond_wait(pool-not_full, pool-lock); } if (pool-shutdown) { pthread_mutex_unlock(pool-lock); return; } pool-tasks[pool-tail].func func; pool-tasks[pool-tail].arg arg; pool-tail (pool-tail 1) % pool-queue_capacity; pool-count; pthread_cond_signal(pool-not_empty); pthread_mutex_unlock(pool-lock); } void pool_destroy(thread_pool *pool) { pthread_mutex_lock(pool-lock); pool-shutdown 1; pthread_cond_broadcast(pool-not_empty); // 唤醒所有 worker pthread_mutex_unlock(pool-lock); for (size_t i 0; i pool-thread_count; i) { pthread_join(pool-threads[i], NULL); } free(pool-tasks); free(pool-threads); }关键点有两个任务在锁外执行t.func(t.arg)放在pthread_mutex_unlock之后。如果把业务逻辑包在锁里就相当于所有线程池工作线程串行执行线程池直接退化极端情况下还会因为任务里恰好提交新任务给同一个池而触发死锁。销毁时用 broadcast所有工作线程都在等not_empty如果只用一个pthread_cond_signal只唤醒一个线程但 shutdown 标志需要所有线程都看到并退出。所以必须在销毁时 broadcast。4.3 线程池的线程数怎么定线程数并不是越多越好。高并发服务里线程数 CPU 核数 I/O 等待占比相关的补偿系数。纯 CPU 密集型任务线程数设成sysconf(_SC_NPROCESSORS_ONLN)附近即可I/O 密集型的比如网络请求里大量 read/write 阻塞等待可以设成核数的 2 到 4 倍。一个简单公式最佳线程数 CPU 核数 × (1 I/O等待耗时 / CPU计算耗时)。不过实际操作中我很少严格按公式更多是先估算再用压测工具把线程数从低往高调观察吞吐量曲线找到平台期。另外要注意排队任务有没有上限。如果队列无界生产者的速度长期超过消费速度内存会被任务对象堆满。绝大多数线上线程池必须给队列设置容量上限满了之后策略由业务决定有人选择直接丢弃配合日志监控有人选择阻塞等待有人选择调用方自己跑一遍任务。5. 线程安全层面锁粒度、原子操作与常见误用5.1 线程安全不是“加了锁就安全”很多人觉得线程安全就是所有共享变量都加锁。实际上锁的正确粒度、持有时间、加锁顺序任何一个搞错都会出问题。加锁顺序不一致是死锁的高发原因线程 A先锁锁1再锁锁2线程 B先锁锁2再锁锁1A 持有锁1 等锁2B 持有锁2 等锁1双双卡死。我在项目里对这种多锁场景的约定是全局规定加锁顺序比如“先队列锁再计数锁”所有人遵守。如果实在避免不了多把锁可以用 trylock失败时主动释放已持有的锁再重试虽然会引入短暂忙等但能避免死锁。5.2 原子操作不能替代所有锁现代 CPU 的 CAScompare-and-swap、原子自增等指令让计数器类操作可以完全无锁#include stdatomic.h atomic_int counter 0; atomic_fetch_add(counter, 1);但原子操作满足的是“单次操作的原子性”不是“一段逻辑的原子性”。比如一个操作是先读变量再决定是否写另一个变量两步之间可能有另一个线程插进来这种复合操作仍然需要锁。典型的例子是在哈希表里先检查 bucket 是否有元素没有就新建节点。检查和一个后续更新操作必须作为一个整体这叫“读-改-写”场景CAS 能处理简单版本复杂一点仍是锁更稳。5.3 单例模式与双重检查锁定的陷阱写单例时很多人用“双重检查锁定”然而纯互斥锁版本有个隐藏的内存可见性问题第一次检查instance NULL时没加锁读到的可能是旧值。严格意义上的安全方案是用 C11 原子操作把指针声明为atomic或直接使用 pthread_oncepthread_once_t once PTHREAD_ONCE_INIT; pthread_mutex_t singleton_lock; void init_singleton(void) { pthread_mutex_init(singleton_lock, NULL); // 其他单例初始化 } void ensure_init(void) { pthread_once(once, init_singleton); }pthread_once保证初始化函数只执行一次并且后续线程能看到完整的初始化结果内存屏障由库内部处理这是 POSIX 层面最省心的答案比自己折腾 DCL 靠谱。5.4 可重入和线程安全是两个维度有一种“加了锁反而出问题”的情况是函数本身就是可重入的但锁不是可重入锁。比如在持有同一个非递归互斥锁的代码路径里再次调用同一个函数而该函数内部又尝试加同一把锁就会死锁。glibc 的pthread_mutex_t默认就是非递归的真要支持同一线程多次加锁得设置PTHREAD_MUTEX_RECURSIVE属性。但递归锁本身往往是设计味道不对的信号先想想能不能拆锁少用递归锁。6. 读写锁读多写少场景的专门优化6.1 为什么需要读写锁互斥锁把“读”和“写”一视同仁两个读者本来可以安全并发也被强制串行。如果一份数据是配置项或者缓存读操作占了 90% 以上用互斥锁浪费严重。读写锁允许读者之间共享写者独占核心 API 是pthread_rwlock_t rwlock; pthread_rwlock_init(rwlock, NULL); // 读路径 pthread_rwlock_rdlock(rwlock); // 读共享数据 pthread_rwlock_unlock(rwlock); // 写路径 pthread_rwlock_wrlock(rwlock); // 修改共享数据 pthread_rwlock_unlock(rwlock);注意一点pthread_rwlock_t的内部实现通常基于原子计数和等待队列性能并不是凭空来的。在多核 CPU 上如果读者们频繁争抢同一个 rwlock 的原子计数缓存行颠簸也会拖慢速度。写者较少时性能提升明显写者比例超过两成读写锁可能还不如轻量互斥锁。6.2 读写锁的优先级策略读写锁一个经典问题是“写饥饿”。如果读者不断涌入写者可能永远等不到锁。POSIX 没有统一规定偏好读还是偏好写由实现决定。Linux glibc 的pthread_rwlock_t默认偏向写者当有写者在等待时新来的读者会被挡在外面直到写者完成。这种机制能避免写者饿死但对读者延迟有影响。判断自己项目里该用哪种策略取决于业务容忍度。如果写操作不能长时间被阻塞比如心跳写状态优先选写者偏好如果写操作更新频率极低、内容不重要读者偏好也许更合适。6.3 读写锁 vs 无锁读有人问既然读多写少能不能直接让读者不加锁不行。不加锁的读者可能看到部分更新的数据比如配置是一个结构体写者先更新字段 A再更新字段 B读者可能在中间读到 A 新但 B 旧的状态。除非你能保证所有读取只依赖一个原子变量否则还是老实加读写锁或者上 RCULinux 内核里有用户态也有liburcu但复杂度更高。7. 常见问题与排查技巧实录7.1 死锁的快速定位方法生产环境死锁是最怕的问题。当程序卡住、CPU 占用率却为 0 时多半是死锁或锁等待。我通常用三步排查gdb -p pid附加到进程thread apply all bt打印所有线程栈看每个线程栈最后几帧找pthread_mutex_lock或pthread_cond_wait对照哪几把锁相互等待如果手头 gdb 不方便pstack或者gdb的批处理模式也够用。关键是确认两个线程锁等待的地址是不是同一把锁。经验线上环境的可观测性比排查技巧更重要。上线前在代码里给每把锁起名字输出日志时带上锁标识一旦死锁日志能直接告诉你“线程A持有锁X等锁Y”。为了省这些日志而后续调死锁完全不值得。7.2 条件变量信号丢失怎么查信号丢失表现为队列明明非空了消费者却一直阻塞在 wait 里程序吞吐量骤降。常见原因有两个生产者 signal 时机在锁外而消费者 wait 前判断条件用的是旧数据消费者 wait 返回后没重新检查条件就消费结果误消费空数据后认为自己消费完毕把 not_full 信号发出去了导致语义混乱排查时先加日志打印队列的 count 变化重点盯住 count 从 0 变 1 的那一瞬间生产者是否调用了 signal以及消费者的判断分支走了哪条。7.3 线程池关闭时的线程泄漏线程池析构如果不把 shutdown 标志置位后 broadcastworker 线程会一直阻塞在pthread_cond_wait导致进程结束时pthread_join挂死。另一个容易忽略的点是任务队列里可能还积压着任务是处理完再关还是直接丢弃必须想清楚并在文档里写明。我的默认选择先广播停止接收新任务然后让 worker 把队列里的任务清空后再退出这样行为可预期不容易在关闭瞬间丢数据。7.4 全局锁导致的性能毛刺有时候功能全对但压测数据很难看所有线程都在等待同一把锁。用perf top能看出热点锁的函数名再用valgrind --toolhelgrind或者-fsanitizethread跑一轮能标出竞争最集中的代码路径。处理方案一般是三条路拆大锁为分段锁比如哈希表的 bucket 各自一把锁把只读数据变成不可变数据读者直接读副本或者用原子操作替代锁。8. 读写锁之外pthread_once 和线程局部存储8.1 pthread_once 的正确使用很多项目需要“保证首次使用时初始化某全局资源”比如日志模块的全局文件句柄。与其写复杂的双重检查不如直接用 pthread_once前面已给出代码。注意 once 控制变量必须是PTHREAD_ONCE_INIT初始化的全局或静态变量绝不能放在栈上。8.2 TLS 线程局部存储减少锁竞争如果每个线程都需要自己的缓存区比如日志缓冲或随机数状态可以用__thread修饰变量static __thread char tls_buffer[4096];TLS 变量每个线程一份天然不需要锁。场景上适合“每个线程独立维护、不跨线程共享”的数据最常见的例子是线程池里每个 worker 自定义的环境信息。如果数据必须跨线程TLS 不适用老老实实用队列。8.3 什么情况需要用户态自旋锁pthread_spinlock_t在锁持有时间极短只有几个原子指令时比互斥锁快因为它不会陷入内核睡眠。但如果临界区超过几十个指令自旋锁会让其他 CPU 核空转反而更浪费。我的使用原则临界区不超过 20~50 个指令且线程数不超过 CPU 核数才考虑自旋。大多数应用场景用 pthread_mutex 就够了内核的 futex 在锁竞争不激烈时开销也很小。9. 我的一点收尾心得把这些东西串起来想线程同步本质上是在回答一个问题多个执行流怎么在共享资源上达成一致。互斥锁解决“同一时间只有一个访问”条件变量解决“状态变化怎么通知到人等”信号量解决“可用资源数量怎么计数”读写锁解决“读多写少怎么偏向共享”。各自适用场景不同没有万能银弹。我自己写并发代码时有个习惯先在注释里写清楚“哪个线程用什么顺序访问哪些共享变量”再动手加锁往往能避免一大半锁顺序问题。调试时优先开 TSan它能揪出让我熬夜到凌晨的数据竞争。这套组合拳下来多线程代码基本能在第一版就站住脚后面再压测调优而不是先爆雷再救火。希望这波分享能给你省下一些和我当年一样惨痛的 debug 时间。