锁的进阶：自旋锁，死锁与条件变量

实验环境：VMware VM（4核，Ubuntu 24.04），g++ 13.2，C++17，glibc 2.39

本文分三部分：自旋锁原理与手写实现、死锁复现与 gdb 定位、条件变量与生产者消费者模型。每部分都有完整代码和实验数据。

一、自旋锁

1.1 硬件原语：Test-and-Set

在实现锁之前，先理解 CPU 提供了什么原子指令。

Test-and-Set（TAS）：一条 CPU 原子指令，读取目标地址的旧值，将其设为 1，返回旧值。整个过程不可打断。

x86 上的 lock bts 指令（Bit Test and Set） lock bts [flag], 0 ; 锁住内存总线，原子操作

Compare-and-Swap（CAS）：比较目标地址的值是否等于预期值，相等则换成新值，返回是否成功。x86 上对应cmpxchg指令。

C++11 将这些硬件原语封装进了标准库。std::atomic_flag是对 TAS 的封装，std::atomic<T>::compare_exchange_strong是对 CAS 的封装。

1.2 手写 spin_lock

自旋锁的核心逻辑极其简单：用一个原子标志位表示"锁是否被占用"，lock()通过 TAS 原子地尝试抢锁，没抢到就继续试；unlock()把标志位清空。

#include<atomic>classspin_lock{std::atomic_flag flag=ATOMIC_FLAG_INIT;public:voidlock(){// test_and_set 返回旧值// flag=false（空闲）→ 设成 true → 返回 false → 拿到锁，退出循环// flag=true（被占）→ 返回 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 VM（4核，Ubuntu 24.04）上运行。

实验

实验统一采用 4 线程各累加一定次数，用std::chrono::steady_clock计时。临界区类型通过调整临界区内的操作来控制：

每种场景分别测试无锁、自旋锁、std::mutex三个版本，记录最终 count 和总耗时。

完整源码如下

#include<iostream>#include<unistd.h>#include<mutex>#include<atomic>#include<thread>#include<vector>#include<chrono>classspin_lock{std::atomic_flag flag=ATOMIC_FLAG_INIT;public:voidlock(){while(flag.test_and_set()){}}voidunlock(){flag.clear();}};intcount=0;spin_lock splk;std::mutex mux;constintPER_THREAD=1000000;constintTHREAD_NUM=4;voidtest_no_lock(){std::vector<std::thread>threads;for(inti=0;i<THREAD_NUM;++i){threads.emplace_back([&]{for(intj=0;j<PER_THREAD;++j)count++;});}for(auto&t:threads)t.join();}voidtest_spin_lock(){std::vector<std::thread>threads;for(inti=0;i<THREAD_NUM;++i){threads.emplace_back([&]{for(intj=0;j<PER_THREAD;++j){splk.lock();count++;splk.unlock();}});}for(auto&t:threads)t.join();}voidtest_mutex(){std::vector<std::thread>threads;for(inti=0;i<THREAD_NUM;++i){threads.emplace_back([&]{for(intj=0;j<PER_THREAD;++j){mux.lock();count++;mux.unlock();}});}for(auto&t:threads)t.join();}template<typenameF>longlongtime_ms(F f){autostart=std::chrono::steady_clock::now();f();autoend=std::chrono::steady_clock::now();returnstd::chrono::duration_cast<std::chrono::milliseconds>(end-start).count();}voidtest_long_critical(){std::vector<std::thread>threads;for(inti=0;i<THREAD_NUM;++i){threads.emplace_back([&](){for(intj=0;j<1000;++j){splk.lock();volatileintx=0;for(intk=0;k<1000000;++k)x+=k;count++;splk.unlock();}});}for(auto&t:threads)t.join();}voidtest_long_critical_mutex(){std::vector<std::thread>threads;for(inti=0;i<THREAD_NUM;++i){threads.emplace_back([&](){for(intj=0;j<1000;++j){mux.lock();volatileintx=0;for(intk=0;k<1000000;++k)x+=k;count++;mux.unlock();}});}for(auto&t:threads)t.join();}intmain(){count=0;autot1=time_ms(test_no_lock);std::cout<<"无锁版 → count="<<count<<" 耗时"<<t1<<"ms\n";count=0;autot2=time_ms(test_spin_lock);std::cout<<"自旋锁版 → count="<<count<<" 耗时"<<t2<<"ms\n";count=0;autot3=time_ms(test_mutex);std::cout<<"mutex版 → count="<<count<<" 耗时"<<t3<<"ms\n";count=0;autot4=time_ms(test_long_critical);std::cout<<"长临界区(自旋锁) → count="<<count<<" 耗时"<<t4<<"ms\n";count=0;autot5=time_ms(test_long_critical_mutex);std::cout<<"长临界区(mutex) → count="<<count<<" 耗时"<<t5<<"ms\n";return0;}

完整代码见 spinlock.cpp。

实验结果

编译运行：

g++-std=c++17-pthreadspinlock.cpp-ospinlock_test ./spinlock_test

▎4核环境（4 线程 × 100 万次）

# ── 实验①：短临界区（4×100 万次 count++）── 无锁版 → count=1619179 耗时37ms 自旋锁版 → count=4000000 耗时339ms mutex版 → count=4000000 耗时115ms # ── 实验②：长临界区·纯计算 10 万次（4×1000）── 长临界区(自旋锁) → count=4000 耗时27ms 长临界区(mutex) → count=4000 耗时45ms # ── 实验③：长临界区·重度计算 100 万次（4×1000）── 长临界区(自旋锁) → count=4000 耗时2805ms 长临界区(mutex) → count=4000 耗时2722ms # ── 实验④：阻塞临界区 usleep(1ms)（仅供参考）── 长临界区(自旋锁) → count=4000 耗时6028ms 长临界区(mutex) → count=4000 耗时5908ms

▎12核环境（4 线程 × 100 万次）

# ── 实验①：短临界区（4×100 万次 count++）── 无锁版 → count=1619179 耗时37ms 自旋锁版 → count=4000000 耗时322ms mutex版 → count=4000000 耗时133ms # ── 实验②：长临界区·重度计算（4×1000）── 长临界区(自旋锁) → count=4000 耗时2778ms 长临界区(mutex) → count=4000 耗时2751ms

结果分析：两套环境（4核 vs 12核），四组实验

▎4核环境（4核VM，4 线程）

实验④单独说明：usleep 触发系统调用，不能代表真正的长临界区行为。4 核下自旋锁 6028ms vs mutex 5908ms，两者均被系统调用开销淹没，仅作参考。

▎12核环境（12核VM，4 线程）

两套环境的数据呈现出一致的模式，说明背后是同一个原因在起作用：

短临界区：自旋锁始终慢于 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 unlock，flag=0 → 核1/2/3的缓存全部失效 核1: test_and_set 成功（flag=1）→ 核2/3的缓存又失效 核2: 从内存重读 flag=1 → 继续自旋 核3: 从内存重读 flag=1 → 继续自旋 每次锁的释放和获取，都引发所有等锁核的缓存失效。

这个开销是自旋锁自有的：锁变量只能在一个时刻被一个核持有，但所有等锁的核都在频繁读它。核越多，缓存一致性流量越大。

而 mutex 不存在这个问题：没抢到锁的线程被 OS 挂起，不产生任何内存访问，只有抢到锁的那个核在正常运行。

长临界区·纯计算（实验②）：自旋锁反超。当临界区是纯计算（10 万次累加），线程一直占着 CPU 不切走。此时自旋锁只在用户态转圈，而 mutex 每次锁竞争都调 futex 系统调用进内核挂起再唤醒——两次上下文切换的开销远超自旋锁的等待。自旋锁快 40%。

临界区极长（实验③）：两者持平。临界区计算时间（3ms 级）远大于锁操作开销，锁的类型不再是考虑因素。

阻塞临界区：usleep 不能用来测长临界区。usleep 本身是系统调用，调用它的线程会主动让出 CPU。此时锁开销被上下文切换淹没了。要测"长临界区"必须用纯计算让线程占着 CPU。

关键结论：自旋锁的性能优势依赖较少的核数竞争——核数越多，cache bouncing 对吞吐的削弱越显著。

二、死锁

自旋锁的特点是"抢不到就一直忙等"，如果线程等不到，就引出了死锁。

2.1 死锁四条件

死锁发生需要同时满足四个条件，缺一不可：互斥、持有并等待、非抢占、循环等待。

预防死锁的本质，就是打破这四个条件中的任意一个。最实用的做法是"破坏循环等待"——固定加锁顺序，所有线程按同样的顺序抢锁。

2.2 死锁复现

最容易写出死锁的方式就是两个线程用相反的顺序抢锁：

#include<mutex>#include<iostream>#include<thread>#include<unistd.h>std::mutex mtx1;std::mutex mtx2;voidfunc1(){// 先锁 mtx1，再锁 mtx2std::lock_guard<std::mutex>lk1(mtx1);sleep(1);std::lock_guard<std::mutex>lk2(mtx2);}voidfunc2(){// 先锁 mtx2，再锁 mtx1std::lock_guard<std::mutex>lk2(mtx2);sleep(1);std::lock_guard<std::mutex>lk1(mtx1);}intmain(){std::threadt1(func1);std::threadt2(func2);t1.join();// 等 func1 结束t2.join();// 等 func2 结束return0;}

编译运行：

g++-std=c++17-pthread-gdeadlock.cpp-odeadlock_test ./deadlock_test&

程序不会结束。控制台没有任何输出，两个线程已经互相锁死了。

2.3 gdb attach：

死锁没有报错、没有崩溃，程序就是不动。需要用 GDB 看一下线程到底卡在哪：

# 先查到进程 IDpsaux|grepdeadlock_test# 用 gdbsudogdb ./deadlock_test-p<PID>

进入 GDB 后，关键命令就两个：

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 1（main）：在join等两个子线程结束
• Thread 2（func2）：卡在futex_wait，正等着锁 mtx1
• Thread 3（func1）：卡在futex_wait，正等着锁 mtx2

(gdb) thread apply all bt Thread 3 (func1): #7 func1() #6 lock_guard<std::mutex> 正在构造 lock_guard（加锁） #1 __lll_lock_wait 底层在等锁 futex_word: <mux2> 等的锁是 mux2 Thread 2 (func2): #7 func2() #6 lock_guard<std::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_guard<std::mutex>lk1(mtx1);// 先 mtx1sleep(1);std::lock_guard<std::mutex>lk2(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_guard

lock_guard不支持手动 unlock/lock，而 wait 内部需要"解锁→休眠→醒来再加锁"这三个步骤。只有unique_lock提供了 unlock() 和 lock() 成员函数，所以条件变量用它更合适。

3.3 生产者消费者

完整代码：

#include<queue>#include<mutex>#include<condition_variable>#include<thread>#include<iostream>std::queue<int>q;std::mutex mtx;std::condition_variable not_empty;std::condition_variable not_full;constintMAX_SIZE=5;constintPRODUCE_NUM=10;voidproducer(intid){for(inti=0;i<PRODUCE_NUM;++i){std::unique_lock<std::mutex>lk(mtx);// 队列满了就等待// 等价于 while(q.size() >= MAX_SIZE) { not_full.wait(lk); }not_full.wait(lk,[]{returnq.size()<MAX_SIZE;});q.push(i);std::cout<<"[P"<<id<<"] produce "<<i<<std::endl;not_empty.notify_one();}}voidconsumer(intid){for(inti=0;i<PRODUCE_NUM;++i){std::unique_lock<std::mutex>lk(mtx);// 队列空了就等待not_empty.wait(lk,[]{return!q.empty();});intval=q.front();q.pop();std::cout<<"[C"<<id<<"] consume "<<val<<std::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::cout<<"ok"<<std::endl;return0;}

编译运行：

g++-std=c++17-pthread-gprod_cons.cpp-oprod_cons ./prod_cons

3.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_SIZE=5）后挂起，消费者连续消费至队空后挂起，双方通过条件变量交替唤醒。4 个线程并发操作 5 容量的队列，未出现溢出或空读，说明条件变量正确保证了每次操作前条件满足。最终生产日志 20 条（2 生产者 × 10）、消费日志 20 条（2 消费者 × 10），数据完整无丢失，所有线程正常结束。

3.6 wait 必须用 while 检查条件

有两种写法：

// 正确写法while(q.size()==MAX_SIZE){not_full.wait(lk);}// 等价正确写法（wait 的谓词重载内置了 while）not_full.wait(lk,[]{returnq.size()<MAX_SIZE;});

但下面这种是错的：

// 错误写法if(q.size()==MAX_SIZE){not_full.wait(lk);}// wait 返回后直接 push——可能队列还是满的！

原因叫做虚假唤醒——线程可能在没有被 notify 的情况下从 wait 返回。这不是 bug，POSIX 和 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_full）+unique_lock实现，4 线程 40 条日志完整无错。

参考资料

• cppreference.comstd::memory_order—— acquire-release / seq_cst 说明
• Linux man pagefutex(2)，man 7 pthreads
• OSTEP 第30章，第31章，第 32 章：条件变量，信号量，常见并发问题

本文是多线程编程系列的第二篇，从自旋锁的硬件原语出发，经过手写实现、benchmark 验证、cache bouncing 分析，再到死锁的复现与排查，最后用条件变量实现了一个完整的生产者消费者模型。由于本人还在学习，如有问题，欢迎评论留言。

系列上一篇：C++多线程入门：创建线程、加锁、计数