实验环境VMware VM4核Ubuntu 24.04g 13.2C17glibc 2.39本文分三部分自旋锁原理与手写实现、死锁复现与 gdb 定位、条件变量与生产者消费者模型。每部分都有完整代码和实验数据。一、自旋锁1.1 硬件原语Test-and-Set在实现锁之前先理解 CPU 提供了什么原子指令。Test-and-SetTAS一条 CPU 原子指令读取目标地址的旧值将其设为 1返回旧值。整个过程不可打断。x86 上的 lock bts 指令Bit Test and Set lock bts [flag], 0 ; 锁住内存总线原子操作Compare-and-SwapCAS比较目标地址的值是否等于预期值相等则换成新值返回是否成功。x86 上对应cmpxchg指令。C11 将这些硬件原语封装进了标准库。std::atomic_flag是对 TAS 的封装std::atomicT::compare_exchange_strong是对 CAS 的封装。1.2 手写 spin_lock自旋锁的核心逻辑极其简单用一个原子标志位表示锁是否被占用lock()通过 TAS 原子地尝试抢锁没抢到就继续试unlock()把标志位清空。#includeatomicclassspin_lock{std::atomic_flag flagATOMIC_FLAG_INIT;public:voidlock(){// test_and_set 返回旧值// flagfalse空闲→ 设成 true → 返回 false → 拿到锁退出循环// flagtrue被占→ 返回 true → 继续循环等待// memory_order_acquire 保证 lock 之后的操作不会重排到 lock 前面while(flag.test_and_set(std::memory_order_acquire)){}}voidunlock(){// memory_order_release 保证 unlock 之前的操作不会重排到 unlock 后面flag.clear(std::memory_order_release);}};整个自旋锁的核心就是test_and_set——它原子地完成读旧值→写 true → 返回旧值三步。两个线程同时调lock()只有一个能拿到false锁空闲另一个拿到true继续转圈。没有系统调用没有内核态切换全程在用户态用一条 CPU 指令完成。这就是自旋锁轻量的本质。1.3 自旋锁 vs 互斥锁理论归理论还需要实际实验去验证。设计一组对照实验分别测试短临界区、长临界区、阻塞临界区IO/sleep 型三种场景下自旋锁和std::mutex的表现。实验环境备注本文所有 benchmark 在 VMware VM4核Ubuntu 24.04上运行。实验实验统一采用 4 线程各累加一定次数用std::chrono::steady_clock计时。临界区类型通过调整临界区内的操作来控制实验编号临界区类型临界区内容内层循环次数重复次数①短临界区无条件竞争count—4×100万②长临界区·纯计算volatile int x; for(k) xk; count10万4×1000③长临界区·重度计算同上100万4×1000④阻塞临界区count; usleep(1ms)—4×1000每种场景分别测试无锁、自旋锁、std::mutex三个版本记录最终 count 和总耗时。完整源码如下#includeiostream#includeunistd.h#includemutex#includeatomic#includethread#includevector#includechronoclassspin_lock{std::atomic_flag flagATOMIC_FLAG_INIT;public:voidlock(){while(flag.test_and_set()){}}voidunlock(){flag.clear();}};intcount0;spin_lock splk;std::mutex mux;constintPER_THREAD1000000;constintTHREAD_NUM4;voidtest_no_lock(){std::vectorstd::threadthreads;for(inti0;iTHREAD_NUM;i){threads.emplace_back([]{for(intj0;jPER_THREAD;j)count;});}for(autot:threads)t.join();}voidtest_spin_lock(){std::vectorstd::threadthreads;for(inti0;iTHREAD_NUM;i){threads.emplace_back([]{for(intj0;jPER_THREAD;j){splk.lock();count;splk.unlock();}});}for(autot:threads)t.join();}voidtest_mutex(){std::vectorstd::threadthreads;for(inti0;iTHREAD_NUM;i){threads.emplace_back([]{for(intj0;jPER_THREAD;j){mux.lock();count;mux.unlock();}});}for(autot:threads)t.join();}templatetypenameFlonglongtime_ms(F f){autostartstd::chrono::steady_clock::now();f();autoendstd::chrono::steady_clock::now();returnstd::chrono::duration_caststd::chrono::milliseconds(end-start).count();}voidtest_long_critical(){std::vectorstd::threadthreads;for(inti0;iTHREAD_NUM;i){threads.emplace_back([](){for(intj0;j1000;j){splk.lock();volatileintx0;for(intk0;k1000000;k)xk;count;splk.unlock();}});}for(autot:threads)t.join();}voidtest_long_critical_mutex(){std::vectorstd::threadthreads;for(inti0;iTHREAD_NUM;i){threads.emplace_back([](){for(intj0;j1000;j){mux.lock();volatileintx0;for(intk0;k1000000;k)xk;count;mux.unlock();}});}for(autot:threads)t.join();}intmain(){count0;autot1time_ms(test_no_lock);std::cout无锁版 → countcount 耗时t1ms\n;count0;autot2time_ms(test_spin_lock);std::cout自旋锁版 → countcount 耗时t2ms\n;count0;autot3time_ms(test_mutex);std::coutmutex版 → countcount 耗时t3ms\n;count0;autot4time_ms(test_long_critical);std::cout长临界区(自旋锁) → countcount 耗时t4ms\n;count0;autot5time_ms(test_long_critical_mutex);std::cout长临界区(mutex) → countcount 耗时t5ms\n;return0;}完整代码见 spinlock.cpp。实验结果编译运行g-stdc17-pthreadspinlock.cpp-ospinlock_test ./spinlock_test▎4核环境4 线程 × 100 万次# ── 实验①短临界区4×100 万次 count── 无锁版 → count1619179 耗时37ms 自旋锁版 → count4000000 耗时339ms mutex版 → count4000000 耗时115ms # ── 实验②长临界区·纯计算 10 万次4×1000── 长临界区(自旋锁) → count4000 耗时27ms 长临界区(mutex) → count4000 耗时45ms # ── 实验③长临界区·重度计算 100 万次4×1000── 长临界区(自旋锁) → count4000 耗时2805ms 长临界区(mutex) → count4000 耗时2722ms # ── 实验④阻塞临界区 usleep(1ms)仅供参考── 长临界区(自旋锁) → count4000 耗时6028ms 长临界区(mutex) → count4000 耗时5908ms▎12核环境4 线程 × 100 万次# ── 实验①短临界区4×100 万次 count── 无锁版 → count1619179 耗时37ms 自旋锁版 → count4000000 耗时322ms mutex版 → count4000000 耗时133ms # ── 实验②长临界区·重度计算4×1000── 长临界区(自旋锁) → count4000 耗时2778ms 长临界区(mutex) → count4000 耗时2751ms结果分析两套环境4核 vs 12核四组实验▎4核环境4核VM4 线程实验场景自旋锁耗时mutex耗时胜负①短临界区339ms115ms自旋锁慢几乎3倍②长临界区·10万次计算27ms45ms自旋锁快 40%③长临界区·100万次计算2805ms2722ms几乎持平实验④单独说明usleep 触发系统调用不能代表真正的长临界区行为。4 核下自旋锁 6028ms vs mutex 5908ms两者均被系统调用开销淹没仅作参考。▎12核环境12核VM4 线程实验场景自旋锁耗时mutex耗时胜负①短临界区322ms133ms自旋锁慢 2.4倍②长临界区·重度计算2778ms2751ms几乎持平两套环境的数据呈现出一致的模式说明背后是同一个原因在起作用短临界区自旋锁始终慢于 mutex。4核和12核下4个线程各自跑在一个核上。每个核都在 while 循环里对同一个 atomic_flag 做 test_and_set——这就触发了cache line bouncing核0: 线程A持有锁跑临界区 核1: 线程B自旋等锁test_and_set(flag) → 读取 flag 核2: 线程C自旋等锁test_and_set(flag) → 读取 flag 核3: 线程D自旋等锁test_and_set(flag) → 读取 flag 线程A unlockflag0 → 核1/2/3的缓存全部失效 核1: test_and_set 成功flag1→ 核2/3的缓存又失效 核2: 从内存重读 flag1 → 继续自旋 核3: 从内存重读 flag1 → 继续自旋 每次锁的释放和获取都引发所有等锁核的缓存失效。这个开销是自旋锁自有的锁变量只能在一个时刻被一个核持有但所有等锁的核都在频繁读它。核越多缓存一致性流量越大。而 mutex 不存在这个问题没抢到锁的线程被 OS 挂起不产生任何内存访问只有抢到锁的那个核在正常运行。长临界区·纯计算实验②自旋锁反超。当临界区是纯计算10 万次累加线程一直占着 CPU 不切走。此时自旋锁只在用户态转圈而 mutex 每次锁竞争都调 futex 系统调用进内核挂起再唤醒——两次上下文切换的开销远超自旋锁的等待。自旋锁快 40%。临界区极长实验③两者持平。临界区计算时间3ms 级远大于锁操作开销锁的类型不再是考虑因素。阻塞临界区usleep 不能用来测长临界区。usleep 本身是系统调用调用它的线程会主动让出 CPU。此时锁开销被上下文切换淹没了。要测长临界区必须用纯计算让线程占着 CPU。关键结论自旋锁的性能优势依赖较少的核数竞争——核数越多cache bouncing 对吞吐的削弱越显著。二、死锁自旋锁的特点是抢不到就一直忙等如果线程等不到就引出了死锁。2.1 死锁四条件死锁发生需要同时满足四个条件缺一不可互斥、持有并等待、非抢占、循环等待。预防死锁的本质就是打破这四个条件中的任意一个。最实用的做法是破坏循环等待——固定加锁顺序所有线程按同样的顺序抢锁。2.2 死锁复现最容易写出死锁的方式就是两个线程用相反的顺序抢锁#includemutex#includeiostream#includethread#includeunistd.hstd::mutex mtx1;std::mutex mtx2;voidfunc1(){// 先锁 mtx1再锁 mtx2std::lock_guardstd::mutexlk1(mtx1);sleep(1);std::lock_guardstd::mutexlk2(mtx2);}voidfunc2(){// 先锁 mtx2再锁 mtx1std::lock_guardstd::mutexlk2(mtx2);sleep(1);std::lock_guardstd::mutexlk1(mtx1);}intmain(){std::threadt1(func1);std::threadt2(func2);t1.join();// 等 func1 结束t2.join();// 等 func2 结束return0;}编译运行g-stdc17-pthread-gdeadlock.cpp-odeadlock_test ./deadlock_test程序不会结束。控制台没有任何输出两个线程已经互相锁死了。2.3 gdb attach死锁没有报错、没有崩溃程序就是不动。需要用 GDB 看一下线程到底卡在哪# 先查到进程 IDpsaux|grepdeadlock_test# 用 gdbsudogdb ./deadlock_test-pPID进入 GDB 后关键命令就两个命令作用thread apply all bt看所有线程的调用栈卡在哪info threads看线程列表总览2.4 输出解读(gdb) info threads Id Target Id Frame * 1 Thread ... deadlock_test __futex_abstimed_wait_common64 ← main 在 join 2 Thread ... deadlock_test futex_wait(mux1) ← 等 mtx1 3 Thread ... deadlock_test futex_wait(mux2) ← 等 mtx2三个线程Thread 1main在join等两个子线程结束Thread 2func2卡在futex_wait正等着锁 mtx1Thread 3func1卡在futex_wait正等着锁 mtx2(gdb) thread apply all bt Thread 3 (func1): #7 func1() #6 lock_guardstd::mutex 正在构造 lock_guard加锁 #1 __lll_lock_wait 底层在等锁 futex_word: mux2 等的锁是 mux2 Thread 2 (func2): #7 func2() #6 lock_guardstd::mutex #1 __lll_lock_wait futex_word: mux1 等的锁是 mux1 Thread 1 (main): #5 main() #4 std::thread::join() 主线程在 join 等子线程GDB 输出的futex_word字段明确指出了每个线程在等哪把锁这是排查死锁最直接的证据。2.5 解决死锁预防死锁最常用的手段固定加锁顺序两个线程都按先 mtx1 后 mtx2的顺序取锁循环等待就不可能存在voidfunc2(){std::lock_guardstd::mutexlk1(mtx1);// 先 mtx1sleep(1);std::lock_guardstd::mutexlk2(mtx2);// 再 mtx2// 没问题了因为 func1 也是这个顺序}其他破坏死锁的手段还有trylock 失败释放重试破坏非抢占条件但可能引入活锁全局预防锁破坏持有并等待一把大锁包住所有抢锁操作无等待数据结构用 CAS 原子指令实现无锁数据结构——破坏互斥条件但实现极复杂三、条件变量死锁是线程互相等对方手里的锁整个程序卡住了。那有没有一种机制让线程在条件不满足时主动休眠条件好了再被唤醒这就是条件变量。自旋锁的问题是等的时候占着 CPU 空转mutex 虽然不空转但只解决互斥、不解决同步3.1 接口条件变量提供了两个核心操作wait让线程在条件不满足时休眠notify_one/notify_all在条件满足时唤醒等待的线程。wait带谓词的版本等价于while(!条件()) wait(lk)——不满足就睡醒来再检查。不带谓词的版本必须手动写 while 循环。wait的内部流程检查条件不满足则解锁 mutex - 线程休眠 - 被 notify 后重新加锁 - 再次检查条件。wait 返回后锁是持有的因为调用方需要接着操作共享数据。3.2 为什么必须用 unique_lock 而不是 lock_guardlock_guard不支持手动 unlock/lock而 wait 内部需要解锁→休眠→醒来再加锁这三个步骤。只有unique_lock提供了 unlock() 和 lock() 成员函数所以条件变量用它更合适。3.3 生产者消费者完整代码#includequeue#includemutex#includecondition_variable#includethread#includeiostreamstd::queueintq;std::mutex mtx;std::condition_variable not_empty;std::condition_variable not_full;constintMAX_SIZE5;constintPRODUCE_NUM10;voidproducer(intid){for(inti0;iPRODUCE_NUM;i){std::unique_lockstd::mutexlk(mtx);// 队列满了就等待// 等价于 while(q.size() MAX_SIZE) { not_full.wait(lk); }not_full.wait(lk,[]{returnq.size()MAX_SIZE;});q.push(i);std::cout[Pid] produce istd::endl;not_empty.notify_one();}}voidconsumer(intid){for(inti0;iPRODUCE_NUM;i){std::unique_lockstd::mutexlk(mtx);// 队列空了就等待not_empty.wait(lk,[]{return!q.empty();});intvalq.front();q.pop();std::cout[Cid] consume valstd::endl;not_full.notify_one();}}intmain(){std::threadp1(producer,1);std::threadp2(producer,2);std::threadc1(consumer,1);std::threadc2(consumer,2);p1.join();p2.join();c1.join();c2.join();std::coutokstd::endl;return0;}编译运行g-stdc17-pthread-gprod_cons.cpp-oprod_cons ./prod_cons3.4 运行结果[P1] produce 0 # P1 连续生产 0~4至队列满-P1 挂起 [P1] produce 1 [P1] produce 2 [P1] produce 3 [P1] produce 4 [C2] consume 0 # C2 连续消费 0~4至队列空-C2 挂起 [C2] consume 1 [C2] consume 2 [C2] consume 3 [C2] consume 4 ... [P1] produce 5~9, [C1] consume 5~9 # C1 消费 P1 [P2] produce 0~4, [C2] consume 0~4 # C2 消费 P2 [P2] produce 5~9, [C1] consume 5~9 # C1 消费 P2 ok # 所有线程正常结束3.5 结果分析运行日志显示生产者连续生产至队列满MAX_SIZE5后挂起消费者连续消费至队空后挂起双方通过条件变量交替唤醒。4 个线程并发操作 5 容量的队列未出现溢出或空读说明条件变量正确保证了每次操作前条件满足。最终生产日志 20 条2 生产者 × 10、消费日志 20 条2 消费者 × 10数据完整无丢失所有线程正常结束。3.6 wait 必须用 while 检查条件有两种写法// 正确写法while(q.size()MAX_SIZE){not_full.wait(lk);}// 等价正确写法wait 的谓词重载内置了 whilenot_full.wait(lk,[]{returnq.size()MAX_SIZE;});但下面这种是错的// 错误写法if(q.size()MAX_SIZE){not_full.wait(lk);}// wait 返回后直接 push——可能队列还是满的原因叫做虚假唤醒——线程可能在没有被 notify 的情况下从 wait 返回。这不是 bugPOSIX 和 C 标准都允许。操作系统为了调度效率偶尔会唤醒等待的线程。如果用的是 if醒来后条件可能仍然不满足但代码已经继续进行了——要么 push 进一个满队列要么 pop 一个空队列。四、总结自旋锁基于 CPU 原子指令 TAS/CAS用户态无系统调用。短临界区多核竞争下因 cache bouncing 比 mutex 慢 2-3 倍长临界区纯计算场景比 mutex 快 40%临界区极长时两者持平。自旋锁的性能优势依赖较少的核数竞争。死锁需同时满足互斥、持有并等待、非抢占、循环等待四个条件。预防核心是破坏循环等待——固定加锁顺序。排查用 gdb attach 后thread apply all bt查看futex_word。条件变量让线程在条件不满足时休眠被唤醒后重新检查条件。wait必须用 while 或谓词重载防止虚假唤醒。生产者消费者双条件变量not_empty not_fullunique_lock实现4 线程 40 条日志完整无错。参考资料cppreference.comstd::memory_order—— acquire-release / seq_cst 说明Linux man pagefutex(2)man 7 pthreadsOSTEP 第30章第31章第 32 章条件变量信号量常见并发问题本文是多线程编程系列的第二篇从自旋锁的硬件原语出发经过手写实现、benchmark 验证、cache bouncing 分析再到死锁的复现与排查最后用条件变量实现了一个完整的生产者消费者模型。由于本人还在学习如有问题欢迎评论留言。系列上一篇C多线程入门创建线程、加锁、计数