写C++并发编程，很多人学完std::thread之后的第一反应是“线程到底开多少个”，真正动手之后才会反应过来：最头疼的不是开线程，而是几十个线程同时去碰同一块数据，怎么保证不出事。这篇就把我在项目里踩过的坑整理一遍，核心其实就两个字——互斥。多线程之间需要共享数据，这是协作的前提；但共享数据不加保护，轻则计数少算几次，重则程序直接挂掉。C++从 C++11 开始提供了一整套同步原语，std::mutex、std::lock_guard、std::scoped_lock，后面又加了std::shared_mutex，目的都是把“同一时刻只能有一个线程碰这块数据”这件事变成规则。这篇文章适合刚学会建线程、却对共享数据没什么把握的人，也适合写过不少并发代码、但还在被数据竞争和死锁困扰的开发者。我会从最小复现例子开始，把锁的选型、粒度、死锁排查全部过一遍。
1. 先搞清楚问题：线程为什么要共享数据、数据竞争从哪来
1.1 线程之间为什么绕不开共享数据
只要线程之间要协作，就一定要交换数据。最简单的例子是任务分发：主线程把任务放进队列，工作线程从队列里取，这个队列就是共享数据。再比如服务器统计请求数，多个线程各自处理完一个请求之后都要把计数器加一。还有一个很常见的场景是缓存热点配置，一个线程在更新，其他线程在读，它们访问的是同一个对象。这还没算复杂业务系统里那种“同一份订单数据被多个模块同时修改”的情况。
共享数据是并发程序里最自然、也最无法回避的东西。理想的设计是尽量让线程各干各的，每个线程持有自己的局部数据，处理完再把结果合并，也就是分治式共享。但真实系统为了实时性和效率，往往还是会把一个可变对象暴露给多个线程。既然绕不开，就必须用同步机制把访问串行化，而最朴素的同步机制就是互斥锁。
1.2 数据竞争为什么是未定义行为
在聊锁之前，先得把“数据竞争”这个概念说清楚。C++ 标准里的定义是：两个线程对同一个内存位置进行访问，其中至少有一个访问是写操作，且这两个访问之间没有强制顺序，就会产生数据竞争。注意这里的“内存位置”可以是同一个整数，同一个对象，也可以是容器里的同一个元素；写操作包括普通赋值和调用非常量成员函数。
数据竞争一旦发生，程序就进入未定义行为。很多人低估了“未定义行为”的杀伤力，以为只是“结果不对”。但在实际项目里它往往比结果不对还难查：编译器看到循环里对普通int做自增，会假设没有其他线程碰它，于是做激进优化，把读、改、写合并，甚至把多个自增重排；硬件缓存也可能让一个核写入的结果在另一个核的缓存里晚很久才可见。一个线程以为自己刚写好的数据，另一个线程在另一个 CPU 核上读到的还是旧值。所以并发代码不能靠运气，必须用同步机制建立明确的先后关系。
1.3 最小复现：十个线程同时给一个 int 自增
先看一段最经典的翻车代码：
#include <iostream> #include <thread> #include <vector> int main() { int counter = 0; std::vector<std::thread> threads; for (int i = 0; i < 10; ++i) { threads.emplace_back([&counter] { for (int j = 0; j < 100000; ++j) { ++counter; } }); } for (auto& t : threads) { t.join(); } std::cout << counter << std::endl; // 期望 1000000 }在单线程环境里，100 万次++counter的结果是确定的。但这里 10 个线程各跑 10 万次，糟糕的时候同一时刻有两三个线程都读到了同一个counter旧值，再分别写回，后写的人把先写的人覆盖掉，总数就少了。++int在汇编层通常不是一条指令，而是 load、add、store 三步，三线程交错就会丢更新。这段程序每次运行结果都可能不一样，哪怕偶尔输出 1000000，也只是巧合，并不是正确。
1.4 先有一个整体选型思路
解决共享数据问题的方案不止互斥锁一种，我一般先做一张小表来选型：
| 方案 | 适用场景 | 主要代价 |
|---|---|---|
std::mutex加 RAII 锁 | 保护结构复杂、需要连续多步操作的数据 | 可能阻塞，有死锁风险 |
std::atomic | 计数器、标志位、简单变量 | 只能针对特定类型和特定操作 |
std::shared_mutex | 读多写少的数据结构 | 写者可能被读者饿死，实现更复杂 |
| 条件变量 + 互斥量 | 线程需要等待共享状态变化 | 必须小心处理等待和唤醒 |
真正的原则是：能减少共享就减少共享，不要一上来就想着加锁。
2. 互斥量：让临界区同一时刻只有一个人进
2.1 裸锁能用，但谁来 unlock？
最原始的写法长这样：
std::mutex m; int counter = 0; void add() { for (int i = 0; i < 100000; ++i) { m.lock(); ++counter; m.unlock(); } }单看逻辑没毛病：想进临界区的人先lock()，干完活unlock()。问题在于，一旦++counter后面的代码抛出异常，或者函数提前return，unlock()就被跳过，互斥量会一直保持锁定，其他线程全部卡在lock()上，程序表现为“假死”。更麻烦的是，如果代码里错误地对同一个互斥量连续调用两次lock()，产生的也是未定义行为。
所以裸锁不是不能用，而是要非常克制。真实代码我更推荐用 RAII 封装，让锁的生命周期和作用域绑定，异常、提前 return 都不会漏解锁。
2.2 lock_guard：作用域替你解锁
std::lock_guard是最简单的 RAII 封装：
std::mutex m; int counter = 0; void add() { std::lock_guard<std::mutex> lock(m); ++counter; }构造lock_guard的时候会调用m.lock()，析构的时候会调用m.unlock()。无论函数是正常走到大括号结束，还是中途抛出异常、提前return，局部对象的析构函数都会被调用，锁一定会释放。临界区就是lock_guard对象生存期覆盖的区域，你不需要数unlock次数。
我一直把它当作默认选择：单把锁、临界区明确、不需要手动解锁，就无脑用lock_guard。
2.3 scoped_lock：一次管住多个互斥量
真正容易踩坑的是需要同时锁多个互斥量的场景，典型例子是两个账户之间转账。假设transfer函数要先锁账户 A 再锁账户 B，另一个函数反向处理，两个线程就可能互相等对方手里的锁，形成死锁。C++17 提供了std::scoped_lock，可以一次接管多个互斥量：
std::mutex m1, m2; void transfer() { std::scoped_lock lock(m1, m2); // 同时持有 m1 和 m2 }scoped_lock内部会采用避免死锁的加锁顺序，在不使用std::lock的情况下也能避免经典死锁。如果是 C++14 及以前，通常写成：
std::lock(m1, m2); std::lock_guard<std::mutex> lock_a(m1, std::adopt_lock); std::lock_guard<std::mutex> lock_b(m2, std::adopt_lock);std::lock负责一次性锁住多个互斥量，adopt_lock告诉lock_guard锁已经拿进来了，析构时只需要释放，不要再加锁。这两行代码写起来容易出错，所以能用scoped_lock就用它。
2.4 unique_lock：需要灵活控制时再换它
lock_guard很稳，但它不提供任何灵活性：你不能手动解锁，也不能在作用域内把锁转移给别人。std::unique_lock补上了这些能力。
典型场景是配合条件变量使用，wait/wait_for要求你传入一个可解锁、可重新加锁的锁，unique_lock就是为这个设计的。另一个场景是想提前结束临界区：数据已经拷贝出来了，后面还有耗时的写日志、回调，你不想继续持锁，就可以在作用域中间手动unlock()，但依然保留 RAII 兜底，防止之后忘记再次解锁。
std::unique_lock<std::mutex> lock(m); data.push_back(x); lock.unlock(); // 提前缩小临界区 do_something_expensive();注意手动unlock()之后，如果后面又访问共享数据，需要重新lock()，否则照样数据竞争。灵活性是双刃剑。
2.5 三个 RAII 封装怎么选
| 工具 | 是否可手动解锁 | 是否支持多锁 | 典型使用场景 |
|---|---|---|---|
std::lock_guard | 否 | 否 | 默认首选，单锁短临界区 |
std::scoped_lock | 否 | 是 | 同时锁多个互斥量，防死锁 |
std::unique_lock | 是 | 可配合 | 条件变量、手动缩临界区 |
我的个人习惯是：单锁用lock_guard，多锁用scoped_lock，遇到条件变量或者明确需要缩小临界区才换unique_lock。别为了“灵活”到处用unique_lock，代码可读性会变差。
3. 真正让代码翻车的不是锁本身，而是锁之外的这些坑
3.1 死锁的经典场面与解决思路
死锁是最容易发现的并发问题，也是最容易写出来的。看这个例子：
std::mutex m1, m2; void thread_a() { std::lock_guard<std::mutex> lock_a(m1); std::lock_guard<std::mutex> lock_b(m2); } void thread_b() { std::lock_guard<std::mutex> lock_b(m2); std::lock_guard<std::mutex> lock_a(m1); }如果线程 A 先拿到m1，线程 B 同时拿到m2，两边都在等对方手里的锁，谁也等不到，程序就卡死了。解决思路有两个：一是让所有线程都按同一个顺序加锁，这样不会出现“我等你，你等我”的环；二是直接用std::scoped_lock同时锁多个互斥量，把选择顺序这件事交给标准库去处理。
死锁发生得非常隐蔽，尤其是代码分布在多个类里的时候。我排查死锁时，第一步不是看代码猜，而是先抓线程栈，看每个线程都阻塞在哪个lock()上，再顺着加锁顺序找环。
3.2 锁的粒度：临界区太大会失去并发，太小会破坏原子性
锁粒度说的是临界区的大小。锁加得太粗，比如整个函数体都放在锁里，哪怕里面有一大半操作根本不涉及共享数据，也会把并发程序活生生变成串行程序，多核优势全没了。锁加得太细，又可能在需要“组合操作”时出问题。
举一个典型场景：更新一个订单对象的价格、数量和更新时间。如果这三个字段各自用一把锁，单独看每个字段都是安全的，但别的线程可能看到“价格已经变了，数量还是旧的”这种中间状态。业务上往往要求这几个字段要么一起变，要么都不变，这就是不变式。正确做法是把“修改这三个字段”定义成一个完整的临界区，而不是颗粒度小到字段级。
判断粒度合不合适，我常用一个原则：临界区里的代码，是否都在操作同一份共享数据？如果操作完成后还要花大量时间做 IO、打印日志、算复杂公式，这些事尽量挪到锁外面。
3.3 把锁藏进接口，别把锁泄露给调用方
很多新手会把std::mutex定义为类的公有成员，或者提供get_mutex()之类的方法，然后让调用方自己加锁。这样做一旦调用方忘记加锁，数据竞争就回来了；更糟的是，调用方可能在一个线程里调用两个成员函数，以为每个成员函数内部各自加锁就够了，结果两个函数合起来需要原子性时，外部又没锁，状态照样坏。
更隐蔽的坑是成员函数返回内部数据的引用或指针。比如：
std::vector<int>& get_items() { return items_; }调用方拿到引用之后，就可以绕过锁直接改items_。我见过太多这样的线上问题：数据竞争不是发生在类内部，而是发生在外部使用返回值的时候。
正确做法是：把需要原子化的操作封装成成员函数，锁完全不对外暴露；要返回数据，就返回拷贝，或者直接返回一个std::optional结果。内部数据坚决不裸奔。
3.4 特殊互斥量：recursive_mutex、timed_mutex，还有读写锁
标准库还提供了几个特殊互斥量。std::recursive_mutex允许同一个线程对同一个互斥量多次加锁，比如一个递归函数中间要反复进入同一个临界区。它听起来很方便，但很容易掩盖设计问题，我一般尽量不用：如果能通过拆分函数避免递归加锁，就不要用递归锁。
std::timed_mutex支持try_lock_for和try_lock_until，可以在等待锁的时候设置超时。适合那些“等不到锁就放弃、做其他事情”的场景，比如网络请求超时重试。要注意的是，锁超时和业务处理超时是两回事，别混着用。
std::shared_mutex则是读写锁：多个读者可以同时读，写者必须独占。适合读多写少的缓存场景。但它也有代价：写者可能长期得不到锁，而且实现更复杂，不是所有平台都优化得很好。
3.5 互斥量和原子操作到底怎么选
很多人会把std::atomic和互斥量对立起来，其实它们是互补的。互斥量适合保护“一个复杂对象”或者“一组操作”，比如队列、链表、多字段状态；原子操作适合保护“单个变量”上的简单操作，比如计数器、标志位、指针切换。
std::atomic<int> counter{0}; counter.fetch_add(1); // 无锁的计数器这个场景如果改用std::mutex，也能把结果算对，但高频自增下会有大量锁竞争，性能差很多。反过来，如果你想保护一个std::map的插入和查找，用原子操作就很难做，因为它不是一个std::atomic能覆盖的类型。
我的经验是：先看共享数据的形态，再选同步工具；不要迷信无锁，也不要什么都上互斥量。
4. 实战：从锁到线程安全队列，直接抄作业
4.1 接口设计原则：所有对外方法都必须自带锁
纸上谈兵讲完了，下面用一个线程安全队列把思路落地。需求很简单：多个生产者线程往队列里 push，多个消费者线程从队列里 pop，队列自身必须保证线程安全。
设计接口时我给自己定了三条规则：
- 所有会读写内部
std::queue的公有方法，内部必须加锁； - 内部数据（包括
std::queue和互斥量）全部私有，不向外暴露引用； - 返回值尽量用值类型，避免返回内部引用。
4.2 完整实现代码
#include <queue> #include <mutex> #include <optional> class SafeQueue { public: void push(int value) { std::lock_guard<std::mutex> lock(mtx_); queue_.push(value); } std::optional<int> try_pop() { std::lock_guard<std::mutex> lock(mtx_); if (queue_.empty()) { return std::nullopt; } int value = queue_.front(); queue_.pop(); return value; } bool empty() const { std::lock_guard<std::mutex> lock(mtx_); return queue_.empty(); } private: std::queue<int> queue_; mutable std::mutex mtx_; };4.3 拆解每个函数：为什么这样写
push很简单，就是加锁后调用底层std::queue::push。try_pop先用empty()判断有没有数据，没有就返回std::nullopt，有就取出front()再pop()。std::optional让“没取到数据”成为一个正常返回值，而不是让调用方去检查一个引用参数有没有被赋值。
empty()被声明为const，但加锁需要修改mtx_，所以mtx_必须是mutable。这一点是新手很容易忽略的：你看着empty是只读操作，但它内部要调lock_guard，本质上改变了互斥量状态。
注意，我在外部同时使用empty()和try_pop()时要小心。if (!queue.empty()) { auto v = queue.try_pop(); }这种先检查再取值的写法，在并发下不是原子的：两个线程可能都检查到非空，然后其中一个pop失败。所以empty()只是给完全无竞争的场景或者调试用，真正判断能否取数据，直接调用try_pop()。
4.4 多线程自测：两个生产者、两个消费者
写一段可编译的测试来验证队列：
#include <atomic> #include <iostream> #include <thread> #include <vector> int main() { SafeQueue queue; std::atomic<int> sum{0}; std::atomic<int> popped{0}; std::atomic<bool> producers_done{false}; std::vector<std::thread> producers; for (int t = 0; t < 2; ++t) { producers.emplace_back([&queue, t] { for (int i = 0; i < 5000; ++i) { queue.push(t * 10000 + i); } }); } std::vector<std::thread> consumers; for (int t = 0; t < 2; ++t) { consumers.emplace_back([&] { while (true) { auto value = queue.try_pop(); if (value.has_value()) { sum += *value; ++popped; continue; } if (producers_done.load() && queue.empty()) { break; } std::this_thread::yield(); } }); } for (auto& thread : producers) { thread.join(); } producers_done = true; for (auto& thread : consumers) { thread.join(); } int expected = 0; for (int t = 0; t < 2; ++t) { for (int i = 0; i < 5000; ++i) { expected += t * 10000 + i; } } std::cout << "popped: " << popped.load() << "\n"; std::cout << "sum: " << sum.load() << "\n"; std::cout << "expected: " << expected << "\n"; }这个测试的核心思路是：生产者和消费者同时跑，消费者不断try_pop，发现队列空并且所有生产者都已经结束，才退出循环。sum和popped都用std::atomic<int>，它们本身也是共享数据，但原子变量在这里是合适的工具。
只要锁实现正确，最后popped一定是 10000，sum一定等于expected。多运行几遍，结果稳定，才说明这个队列基本可用。
4.5 这个队列还能怎么改进
当前版本是忙等轮询，消费者在队列空时会一直yield()，浪费 CPU。真实项目通常会改成阻塞式：消费者在队列空时睡在条件变量上，生产者 push 完数据后唤醒一个消费者。这就需要std::condition_variable和std::unique_lock配合，是下一阶段要讲的内容。
如果并发量特别大，单把锁也可能成为瓶颈。后续可以考虑分片锁、多队列，或者用无锁队列。但这些都是优化，不是第一要务。先把一把锁用对，再谈优化。
5. 常见问题与排查技巧实录
5.1 症状一：结果偶发不对，程序还会崩溃
最典型的表现就是前面那个自增例子：程序大部分时间能跑通，偶尔输出少一截，甚至直接在free或析构时崩溃。这种问题八成是数据竞争。
排查数据竞争最好不要靠肉眼盯代码，工具比我靠谱。GCC 和 Clang 都支持ThreadSanitizer，编译时加几个选项就能跑：
g++ -std=c++17 -fsanitize=thread -g main.cpp -o main -pthread ./mainThreadSanitizer 会在检测到数据竞争时直接打印访问位置和线程栈，顺着它给的线索改，通常能很快定位。缺点是会显著拖慢程序，只建议在调试阶段开。
5.2 症状二：程序卡死，所有线程都在等锁
死锁的表现通常是程序没有任何输出，CPU 占用极低，像卡住了一样。这时先别急着改代码，用调试器看线程栈：
gdb -p <pid>然后在 gdb 里执行：
thread apply all bt每个线程的堆栈都会打印出来。如果看到两个线程分别阻塞在不同互斥量的lock()上，而它们又互相持有对方想要的锁，死锁就实锤了。
解决方向我已经在前面说过：统一加锁顺序，或者用std::scoped_lock。还有一个笨办法，把代码里所有lock()临时替换成try_lock_for()，超时后打印日志，能很快发现是谁在长时间等锁。
5.3 症状三：加锁正确了，程序却慢得离谱
加锁加对不代表性能没问题。如果线程多但临界区又长，其他线程大部分时间都在排队等锁，多核并发反而比单线程还慢。我遇到过最夸张的情况是，一个线程在锁里写日志，写磁盘花了 100 毫秒，其他 15 个线程全部排队。
排查性能热点用perf top就能看出来，也可以给临界区前后加计时点。解决思路通常有三个：把耗时的 IO 和计算挪到锁外；用读写锁允许并发读；或者把大共享数据结构拆成多个独立的小结构，让线程尽量访问不同的片段。
5.4 问题速查表
| 现象 | 可能原因 | 快速排查与解决 |
|---|---|---|
| 随机丢数据、多跑几次结果不同 | 数据竞争 | 开 ThreadSanitizer，加锁或改用原子变量 |
| 程序卡死，CPU 占用低 | 死锁 | 看线程栈，统一加锁顺序，用 scoped_lock |
| 已经加锁，但程序还是很慢 | 锁竞争严重、临界区过长 | 缩小临界区，将 IO/计算放锁外 |
| 修改后其他线程长时间看不到 | 缺少内存同步 | 用互斥锁或原子操作建立 happens-before |
| 持有锁时调用外部回调导致卡死 | 回调里反向请求同一个锁 | 内部拷贝数据，解锁后再回调 |
5.5 最后几点避坑心得
写多线程代码，我自己的习惯是先画一张“谁在什么条件下访问哪些变量”的小表，不需要很正式，列清楚写操作和读操作就行。多线程代码最怕的不是写得不够聪明，而是写得不够平凡：能用lock_guard就用lock_guard，能别返回内部引用就别返回，能封装到类内部就别把锁暴露出去。
还有一点很重要：不要在持有锁的时候调用那些可能长时间阻塞或者会再锁别的互斥量的外部函数。你永远不知道调用方会在哪里遇到另一把锁，一旦形成依赖环，就是死锁。我宁愿先拷贝数据，解锁，再去做后面的操作。
最后再说一个观察：实际项目中，真正复杂的问题很少是“我不会用锁”，而是“哪些操作组合起来必须是一个原子操作”。把业务上的不变式想清楚，再动手写锁，比什么都重要。
