锁的进阶：从自旋锁到手写实现，再到死锁与条件变量

实验环境：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>

class spin_lock{
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
    void lock(){
        // test_and_set 返回旧值
        // flag=false（空闲）→ 设成 true → 返回 false → 拿到锁，退出循环
        // flag=true（被占）→ 返回 true → 继续循环等待
        // memory_order_acquire 保证 lock 之后的操作不会重排到 lock 前面
        while(flag.test_and_set(std::memory_order_acquire)){
        }
    }

    void unlock(){
        // 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>

class spin_lock{
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
    void lock(){
        while(flag.test_and_set()){}
    }
    void unlock(){
        flag.clear();
    }
};

int count=0;
spin_lock splk;
std::mutex mux;
const int PER_THREAD=1000000;
const int THREAD_NUM=4;

void test_no_lock(){
    std::vector<std::thread> threads;
    for(int i=0;i<THREAD_NUM;++i){
        threads.emplace_back([&]{
            for(int j=0;j<PER_THREAD;++j) count++;
        });
    }
    for(auto &t:threads) t.join();
}

void test_spin_lock(){
    std::vector<std::thread> threads;
    for(int i=0;i<THREAD_NUM;++i){
        threads.emplace_back([&]{
            for(int j=0;j<PER_THREAD;++j){
                splk.lock(); count++; splk.unlock();
            }
        });
    }
    for(auto &t:threads) t.join();
}

void test_mutex(){
    std::vector<std::thread> threads;
    for(int i=0;i<THREAD_NUM;++i){
        threads.emplace_back([&]{
            for(int j=0;j<PER_THREAD;++j){
                mux.lock(); count++; mux.unlock();
            }
        });
    }
    for(auto &t:threads) t.join();
}

template<typename F>
long long time_ms(F f){
    auto start = std::chrono::steady_clock::now();
    f();
    auto end = std::chrono::steady_clock::now();
    return std::chrono::duration_cast<std::chrono::milliseconds>(end-start).count();
}

void test_long_critical(){
    std::vector<std::thread> threads;
    for(int i=0;i<THREAD_NUM;++i){
        threads.emplace_back([&](){
            for(int j=0;j<1000;++j){
                splk.lock();
                volatile int x=0;
                for(int k=0;k<1000000;++k) x+=k;
                count++;
                splk.unlock();
            }
        });
    }
    for(auto &t:threads) t.join();
}

void test_long_critical_mutex(){
    std::vector<std::thread> threads;
    for(int i=0;i<THREAD_NUM;++i){
        threads.emplace_back([&](){
            for(int j=0;j<1000;++j){
                mux.lock();
                volatile int x=0;
                for(int k=0;k<1000000;++k) x+=k;
                count++;
                mux.unlock();
            }
        });
    }
    for(auto &t:threads) t.join();
}

int main(){
    count = 0;
    auto t1 = time_ms(test_no_lock);
    std::cout << "无锁版  → count=" << count << " 耗时" << t1 << "ms\n";

    count = 0;
    auto t2 = time_ms(test_spin_lock);
    std::cout << "自旋锁版 → count=" << count << " 耗时" << t2 << "ms\n";

    count = 0;
    auto t3 = time_ms(test_mutex);
    std::cout << "mutex版  → count=" << count << " 耗时" << t3 << "ms\n";

    count = 0;
    auto t4 = time_ms(test_long_critical);
    std::cout << "长临界区(自旋锁) → count=" << count << " 耗时" << t4 << "ms\n";

    count = 0;
    auto t5 = time_ms(test_long_critical_mutex);
    std::cout << "长临界区(mutex)  → count=" << count << " 耗时" << t5 << "ms\n";
    return 0;
}

完整代码见 spinlock.cpp。

实验结果

编译运行：

g++ -std=c++17 -pthread spinlock.cpp -o spinlock_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;

void func1(){
	// 先锁 mtx1，再锁 mtx2
	std::lock_guard<std::mutex> lk1(mtx1);
	sleep(1);                    
	std::lock_guard<std::mutex> lk2(mtx2);  
}

void func2(){
	// 先锁 mtx2，再锁 mtx1 
	std::lock_guard<std::mutex> lk2(mtx2);
	sleep(1);                   
	std::lock_guard<std::mutex> lk1(mtx1);  
}

int main(){
	std::thread t1(func1);
	std::thread t2(func2);
	t1.join();   // 等 func1 结束
	t2.join();   // 等 func2 结束
	return 0;
}

编译运行：

g++ -std=c++17 -pthread -g deadlock.cpp -o deadlock_test
./deadlock_test &

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

2.3 gdb attach：

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

# 先查到进程 ID
ps aux | grep deadlock_test

# 用 gdb 
sudo gdb ./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"的顺序取锁，循环等待就不可能存在：

void func2(){
	std::lock_guard<std::mutex> lk1(mtx1);  // 先 mtx1
	sleep(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;   
const int MAX_SIZE = 5;
const int PRODUCE_NUM = 10;

void producer(int id){
    for(int i = 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, []{ return q.size() < MAX_SIZE; });

        q.push(i);
        std::cout << "[P" << id << "] produce " << i << std::endl;

        not_empty.notify_one();  
    }
}

void consumer(int id){
    for(int i = 0; i < PRODUCE_NUM; ++i){
        std::unique_lock<std::mutex> lk(mtx);
        // 队列空了就等待
        not_empty.wait(lk, []{ return !q.empty(); });

        int val = q.front();
        q.pop();
        std::cout << "[C" << id << "] consume " << val << std::endl;

        not_full.notify_one();  
    }
}

int main(){
    std::thread p1(producer, 1);
    std::thread p2(producer, 2);
    std::thread c1(consumer, 1);
    std::thread c2(consumer, 2);
    p1.join(); p2.join(); c1.join(); c2.join();
    std::cout << "ok" << std::endl;
    return 0;
}

编译运行：

g++ -std=c++17 -pthread -g prod_cons.cpp -o prod_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, []{ return q.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.com std::memory_order—— acquire-release / seq_cst 说明
• Linux man page futex(2)，man 7 pthreads
• OSTEP 第30章，第31章，第 32 章：条件变量，信号量，常见并发问题

---

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

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