前言多线程同步的核心问题只有一个多个线程同时读写同一块内存时如何保证结果正确。C 内存模型给出的答案是happens-before 关系——如果线程 A 的写操作 happens-before 线程 B 的读操作B 就一定能看到 A 写下的值如果没有这个关系两个线程对同一块非原子内存的并发访问至少一个是写就是数据竞争data race而数据竞争在标准里是未定义行为Undefined BehaviorUB标准不保证任何结果。这里有两个流传极广的误解要先纠正。第一个是加个volatile就能多线程共享变量了。这是错的volatile既不提供原子性也不建立任何跨线程的 happens-before 关系它只能阻止编译器把访问优化掉。第二个是数据竞争最多只是读到旧值。实际上 UB 意味着优化器可以基于不存在数据竞争的假设做重排产生和源码逻辑完全对不上的行为。本文按互斥量、条件变量、读写锁、原子与内存序四条路线讲清各类同步设施的语义边界。代码以 C17 为基准在 GCC 13 / Clang 17 / MSVC 19.3x 上均可编译涉及 C20 的特性会单独标注。一、互斥量从 lock_guard 到 scoped_lock最基础的同步手段是互斥量mutex。要点是必须用 RAII 包装类管理加解锁手动lock()/unlock()会在异常路径或提前return时漏掉解锁从而把后续所有线程永久堵死。#include mutex class Counter { public: void add(long n) { std::lock_guardstd::mutex lock(mtx_); value_ n; } long value() const { std::lock_guardstd::mutex lock(mtx_); return value_; } private: mutable std::mutex mtx_; // const 成员函数里也要加锁所以是 mutable long value_ 0; };让四个线程各调用add(1)一万次结果必定是 40000如果去掉lock_guard结果不可预期——那已经是数据竞争UB不是少算几次。两个细节mtx_必须声明为mutable否则const成员函数value()无法调用lock()lock()是非常量成员函数std::lock_guard在 C17 里有推导指引deduction guide省略模板参数的写法也合法。std::unique_lock比lock_guard稍重但换来可以中途解锁、可以移动、可以交给条件变量使用的能力。只在一段作用域里加锁就用lock_guard需要和条件变量配合或提前解锁时才用unique_lock。多个互斥量同时加锁最容易踩死锁线程 1 先锁 A 再锁 B线程 2 先锁 B 再锁 A两边各持一把、互等对方。C17 的std::scoped_lock一次把多个锁一起拿到内部采用能避免死锁的算法#include mutex struct Account { long balance 0; }; void transfer(Account from, Account to, long amount, std::mutex m1, std::mutex m2) { std::scoped_lock lock(m1, m2); // C17同时锁住内部避免死锁 from.balance - amount; to.balance amount; }C17 之前要写成std::lock(m1, m2);再加两个带std::adopt_lock的std::lock_guardadopt_lock的意思是锁已经被拿走了别再锁一次。std::scoped_lock也支持单参数用法这时它等价于lock_guard。二、条件变量为什么要用谓词循环互斥量解决不能同时访问条件变量解决等某个条件成立再继续。条件变量必须配合std::unique_lockstd::mutex使用它需要在等待期间释放锁、被唤醒后再重新获取而lock_guard没有unlock()做不到这件事。#include condition_variable #include mutex #include queue #include utility template typename T class BlockingQueue { public: void push(T value) { { std::lock_guardstd::mutex lock(mtx_); queue_.push(std::move(value)); } // 先解锁 cv_.notify_one(); // 再通知被唤醒的线程能立刻拿到锁 } T pop() { std::unique_lockstd::mutex lock(mtx_); cv_.wait(lock, [this] { return !queue_.empty(); }); // 谓词形式 T value std::move(queue_.front()); queue_.pop(); return value; } private: std::mutex mtx_; std::condition_variable cv_; std::queueT queue_; };cv_.wait(lock, pred)的语义是一个循环先检查pred()为真就返回否则原子地释放锁并阻塞被唤醒后再重新拿锁、再检查。标准明确允许虚假唤醒spurious wakeup——即使没有线程调用notifywait也可能返回。因此下面这种不带谓词的写法是错的// ❌ 虚假唤醒后会带着空队列继续执行 std::unique_lockstd::mutex lock(mtx_); cv_.wait(lock); T value queue_.front(); // 队列为空时调用 front() 是 UB带谓词的wait就等价于手写while (!pred()) { cv_.wait(lock); }。push里先解锁再notify_one是常见的效率写法持锁通知时被唤醒的线程会立刻尝试拿锁拿不到就再次阻塞多了一次无谓的上下文切换。另外std::condition_variable只能配std::unique_lockstd::mutexstd::condition_variable_any可配任意满足锁概念的锁但开销也更大。三、读写锁读多写少时的选择当临界区大部分操作是只读、只有少部分是写时用普通互斥量会让所有读操作也串行化。C17 引入了std::shared_mutex读写锁#include map #include shared_mutex #include string #include utility class Config { public: std::string get(const std::string key) const { std::shared_lockstd::shared_mutex lock(mtx_); // 共享读锁 auto it map_.find(key); return it map_.end() ? std::string{} : it-second; } void set(const std::string key, std::string value) { std::unique_lockstd::shared_mutex lock(mtx_); // 独占写锁 map_[key] std::move(value); } private: mutable std::shared_mutex mtx_; std::mapstd::string, std::string map_; };std::shared_mutex要求包含头文件shared_mutexC17 起提供。读锁用std::shared_lockC14 引入的模板C17 起可配合shared_mutex写锁用std::unique_lock或std::lock_guard。必须提醒的是读写锁不一定比普通互斥量快它要维护读者计数和写者等待状态加解锁的固定开销比std::mutex高是否值得要用实测说话。另外标准只定义了共享/独占的接口约定调度策略属于实现定义libstdc、libc、MSVC STL 各不相同不要依赖写请求优先。四、原子操作与内存序另一条路互斥量靠排他来保证正确性std::atomic靠禁止数据竞争来保证正确性。对于单个变量上的简单操作原子操作通常比加锁轻量而且不会阻塞。先看内存序memory order的语义C 提供六种内存序语义要点典型用途memory_order_relaxed只保证该变量上的操作是原子的、且同一变量上的修改有单一总序不建立任何跨线程的同步单纯计数、互不依赖的统计量memory_order_acquire用于读操作之后的读写不能被重排到该操作之前读取数据已就绪标志memory_order_release用于写操作之前的读写不能被重排到该操作之后写入数据已就绪标志memory_order_acq_rel读改写操作兼具两者如fetch_add、成功的 CAS无锁结构里的 CASmemory_order_seq_cst在上述基础上再加一个所有线程一致的全局总序这是各成员函数的默认值不确定时先用它memory_order_consume依赖链语义各家实现基本等同于acquire不推荐使用基本不用最经典的发布-订阅模式如下。data是一个普通int但在release/acquire的配合下并发访问不再是数据竞争#include atomic #include cassert #include thread std::atomicbool ready{false}; int data 0; void producer() { data 42; // 普通写 ready.store(true, std::memory_order_release); // 保证 data 42 在 store 之前完成 } void consumer() { while (!ready.load(std::memory_order_acquire)) { // 自旋等待 } // 这里的 data 一定等于 42acquire 与 release 建立了 happens-before assert(data 42); }两个线程分别跑producer和consumer最后join即可。如果把ready的 store 换成memory_order_relaxeddata 42和ready.store(true)之间就没有任何顺序保证编译器或 CPU 都可以把ready的写提前。此时consumer读到的data是不是 42 完全不确定而且data的这次并发读写本身就是数据竞争是 UB——不是读到 0而是标准不保证任何行为。memory_order_relaxed也不是没用。上面的Counter如果换成std::atomiclong value_value_.fetch_add(n, std::memory_order_relaxed)就完全够用每次fetch_add都是原子的最终总和与执行顺序无关不需要任何跨线程的顺序约束。4.1 用 atomic_flag 实现自旋锁#include atomic class SpinLock { public: void lock() { while (flag_.test_and_set(std::memory_order_acquire)) { } // 自旋等待 } void unlock() { flag_.clear(std::memory_order_release); } private: std::atomic_flag flag_ ATOMIC_FLAG_INIT; // C17 的写法 };std::atomic_flag是唯一的保证无锁的原子类型C17 下只有test_and_set和clear两个操作test、wait、notify_one都是 C20 才加入的并且必须用ATOMIC_FLAG_INIT初始化。这里的acquire/release配对是必需的前者保证拿到锁之后的读写不会跑到加锁之前后者保证解锁之前的读写不会跑到解锁之后。自旋锁只适合临界区极短、线程数不超过核心数的场景。4.2 无锁栈与 ABA 问题无锁lock-free结构用 CAScompare-and-swap在 C 里是compare_exchange_weak/compare_exchange_strong替换加锁。下面是一个经典的 Treiber 栈#include atomic template typename T class LockFreeStack { struct Node { T value; Node* next; }; public: bool pop(T out) { Node* old_head head_.load(std::memory_order_acquire); while (old_head ! nullptr !head_.compare_exchange_weak(old_head, old_head-next, std::memory_order_acquire, std::memory_order_acquire)) { // 失败时 old_head 被更新重试 } if (old_head nullptr) { return false; } out old_head-value; // 注意这里故意不 delete见下文对 ABA 与内存回收的说明 return true; } private: std::atomicNode* head_{nullptr}; };push用同样的 CAS 循环把新节点挂到head_上。compare_exchange_weak的签名是期望值按引用传入、目标值按值传入失败时会把当前实际值写回第一个参数。weak版本允许伪失败即使值相等也可能返回 false在 LL/SC 架构上很常见所以必须放在循环里。这个结构有一个著名的缺陷ABA 问题。pop的第一步是读head_第二步才 CAS。在这两步之间head_可能被别的线程改成了别的值、又改回来初始: head_ - A - B - C T1: 读到 old_head A尚未执行 CAS被挂起 T2: pop A、pop B然后 push 了一个新节点 A2A2 恰好被分配在 A 原来的地址上 T1: 恢复执行比较 head_ A —— 地址相同比较成立 于是把 head_ 设为 old_head-next问题就在这里old_head-next读的是当初那个节点对象的字段。如果这块内存在 T2 手里被释放又被复用T1 读到的就是 A2 的nexthead_于是指向一个错误的地址B、C 这类节点就此从栈上消失。而如果 T2 已经delete了 AT1 再去读A-next就是释放后使用use-after-free属于 UB。解决 ABA 的常见手段有三类一是给指针带上版本号tagged pointer用双字宽度的 CAS 一起比较指针和版本号但这要求目标平台支持无锁的双字 CAS二是风险指针hazard pointer或基于纪元的回收epoch-based reclamation在确认没有线程引用该节点后再释放三是换用 C20 的std::atomicstd::shared_ptrT需要 C20靠引用计数规避节点被提前释放——C17 里的替代方案是 C11 就有的std::atomic_load/std::atomic_store自由函数作用于shared_ptr。需要把话说透无锁不等于快也不等于简单。上面这个pop只是教学骨架——不回收内存会泄漏、不处理 ABA也没解决何时能安全删除节点这个真正困难的问题。生产代码里一个加std::mutex的栈在绝大多数场景下都更短、更对、更容易维护。常见坑点场景❌ 错误写法✅ 正确写法用volatile同步volatile bool stop; while (!stop) {}std::atomicbool stopload(std::memory_order_acquire)条件变量不写谓词cv.wait(lock);后直接读队列配lock_guard更是编译不过cv.wait(lock, []{ return !q.empty(); });锁用std::unique_lockstd::mutex忘了初始化atomic_flagstd::atomic_flag f;C17状态不确定std::atomic_flag f ATOMIC_FLAG_INIT;两个锁加锁顺序相反lock(a); lock(b);与lock(b); lock(a);并存std::scoped_lock lock(a, b);C17只靠relaxed传递数据用relaxed写就绪标志再读普通变量标志用release/acquire配对weakCAS 不放在循环里if (!head_.compare_exchange_weak(...)) return false;用while (...)循环容忍伪失败无锁结构直接delete节点CAS 成功后立即delete old_head;而其他线程仍持有该指针用 hazard pointer / 纪元回收 / 引用计数或改用互斥量再补一条容易被忽略的不要把std::mutex的地址跨线程传递后再依赖它的生命周期。互斥量本身也是对象如果它随某个对象一起析构了而另一个线程还在lock()它那就是访问已销毁对象属于 UB。总结需求推荐设施关键约束保护一段临界区std::lock_guardstd::mutex全程 RAII不手动lock/unlock需要中途解锁或配合条件变量std::unique_lock比lock_guard多一个状态标志略重同时锁多个互斥量std::scoped_lockC17内部算法避免死锁等待条件成立std::condition_variable 谓词必须容忍虚假唤醒读多写少std::shared_mutexC17std::shared_lock是否更快需实测写者优先无保证单变量计数std::atomicrelaxed只需原子性无顺序要求发布数据给其他线程release写 acquire读缺一不可否则是数据竞争UB无锁结构CAS 循环必须处理 ABA 与节点回收两件事选同步方式的顺序建议是先用互斥量把逻辑写对确认它确实是瓶颈之后再考虑替换。无锁代码要额外承担 ABA、内存回收、伪失败重试和改一行就出错的维护成本真正需要它的场景是临界区短到加锁开销已占大头、并发度极高的热点路径。在那之前把锁的粒度缩小、把临界区里的耗时操作I/O、内存分配、日志挪出去往往收益更大也更安全。