企业官网建设流程全解析

前言

多线程同步的核心问题只有一个：多个线程同时读写同一块内存时，如何保证结果正确。C++ 内存模型给出的答案是"happens-before 关系"——如果线程 A 的写操作 happens-before 线程 B 的读操作，B 就一定能看到 A 写下的值；如果没有这个关系，两个线程对同一块非原子内存的并发访问（至少一个是写）就是数据竞争（data race），而数据竞争在标准里是未定义行为（Undefined Behavior，UB），标准不保证任何结果。

这里有两个流传极广的误解要先纠正。第一个是"加个volatile就能多线程共享变量了"。这是错的：volatile既不提供原子性，也不建立任何跨线程的 happens-before 关系，它只能阻止编译器把访问优化掉。第二个是"数据竞争最多只是读到旧值"。实际上 UB 意味着优化器可以基于"不存在数据竞争"的假设做重排，产生和源码逻辑完全对不上的行为。

本文按"互斥量、条件变量、读写锁、原子与内存序"四条路线讲清各类同步设施的语义边界。代码以 C++17 为基准，在 GCC 13 / Clang 17 / MSVC 19.3x 上均可编译，涉及 C++20 的特性会单独标注。

一、互斥量：从 lock_guard 到 scoped_lock

最基础的同步手段是互斥量（mutex）。要点是必须用 RAII 包装类管理加解锁，手动lock()/unlock()会在异常路径或提前return时漏掉解锁，从而把后续所有线程永久堵死。

#include <mutex> class Counter { public: void add(long n) { std::lock_guard<std::mutex> lock(mtx_); value_ += n; } long value() const { std::lock_guard<std::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在 C++17 里有推导指引（deduction guide），省略模板参数的写法也合法。

std::unique_lock比lock_guard稍重，但换来可以中途解锁、可以移动、可以交给条件变量使用的能力。只在一段作用域里加锁就用lock_guard，需要和条件变量配合或提前解锁时才用unique_lock。

多个互斥量同时加锁最容易踩死锁：线程 1 先锁 A 再锁 B，线程 2 先锁 B 再锁 A，两边各持一把、互等对方。C++17 的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); // C++17；同时锁住，内部避免死锁 from.balance -= amount; to.balance += amount; }

C++17 之前要写成std::lock(m1, m2);再加两个带std::adopt_lock的std::lock_guard，adopt_lock的意思是"锁已经被拿走了，别再锁一次"。std::scoped_lock也支持单参数用法，这时它等价于lock_guard。

二、条件变量：为什么要用谓词循环

互斥量解决"不能同时访问"，条件变量解决"等某个条件成立再继续"。条件变量必须配合std::unique_lock<std::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_guard<std::mutex> lock(mtx_); queue_.push(std::move(value)); } // 先解锁 cv_.notify_one(); // 再通知：被唤醒的线程能立刻拿到锁 } T pop() { std::unique_lock<std::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::queue<T> queue_; };

cv_.wait(lock, pred)的语义是一个循环：先检查pred()，为真就返回；否则原子地释放锁并阻塞，被唤醒后再重新拿锁、再检查。标准明确允许虚假唤醒（spurious wakeup）——即使没有线程调用notify，wait也可能返回。因此下面这种不带谓词的写法是错的：

// ❌ 虚假唤醒后会带着空队列继续执行 std::unique_lock<std::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_lock<std::mutex>，std::condition_variable_any可配任意满足锁概念的锁，但开销也更大。

三、读写锁：读多写少时的选择

当临界区大部分操作是只读、只有少部分是写时，用普通互斥量会让所有读操作也串行化。C++17 引入了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_lock<std::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_lock<std::shared_mutex> lock(mtx_); // 独占（写）锁 map_[key] = std::move(value); } private: mutable std::shared_mutex mtx_; std::map<std::string, std::string> map_; };

std::shared_mutex要求包含头文件<shared_mutex>，C++17 起提供。读锁用std::shared_lock（C++14 引入的模板，C++17 起可配合shared_mutex），写锁用std::unique_lock或std::lock_guard。

必须提醒的是：读写锁不一定比普通互斥量快，它要维护读者计数和写者等待状态，加解锁的固定开销比std::mutex高，是否值得要用实测说话。另外标准只定义了共享/独占的接口约定，调度策略属于实现定义，libstdc++、libc++、MSVC STL 各不相同，不要依赖"写请求优先"。

四、原子操作与内存序：另一条路

互斥量靠"排他"来保证正确性，std::atomic靠"禁止数据竞争"来保证正确性。对于单个变量上的简单操作，原子操作通常比加锁轻量，而且不会阻塞。

先看内存序（memory order）的语义，C++ 提供六种：

最经典的"发布-订阅"模式如下。data是一个普通int，但在release/acquire的配合下，并发访问不再是数据竞争：

#include <atomic> #include <cassert> #include <thread> std::atomic<bool> 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 一定等于 42：acquire 与 release 建立了 happens-before assert(data == 42); }

两个线程分别跑producer和consumer，最后join即可。如果把ready的 store 换成memory_order_relaxed，data = 42和ready.store(true)之间就没有任何顺序保证，编译器或 CPU 都可以把ready的写提前。此时consumer读到的data是不是 42 完全不确定，而且data的这次并发读写本身就是数据竞争，是 UB——不是"读到 0"，而是"标准不保证任何行为"。

memory_order_relaxed也不是没用。上面的Counter如果换成std::atomic<long> 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; // C++17 的写法 };

std::atomic_flag是唯一的"保证无锁"的原子类型，C++17 下只有test_and_set和clear两个操作（test、wait、notify_one都是 C++20 才加入的），并且必须用ATOMIC_FLAG_INIT初始化。这里的acquire/release配对是必需的：前者保证拿到锁之后的读写不会跑到加锁之前，后者保证解锁之前的读写不会跑到解锁之后。自旋锁只适合临界区极短、线程数不超过核心数的场景。

4.2 无锁栈与 ABA 问题

无锁（lock-free）结构用 CAS（compare-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::atomic<Node*> 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 了一个新节点 A2，A2 恰好被分配在 A 原来的地址上 T1: 恢复执行，比较 head_ == A —— 地址相同，比较成立 于是把 head_ 设为 old_head->next

问题就在这里：old_head->next读的是当初那个节点对象的字段。如果这块内存在 T2 手里被释放又被复用，T1 读到的就是 A2 的next，head_于是指向一个错误的地址，B、C 这类节点就此从栈上"消失"。而如果 T2 已经delete了 A，T1 再去读A->next就是释放后使用（use-after-free），属于 UB。

解决 ABA 的常见手段有三类：一是给指针带上版本号（tagged pointer），用双字宽度的 CAS 一起比较指针和版本号，但这要求目标平台支持无锁的双字 CAS；二是风险指针（hazard pointer）或基于纪元的回收（epoch-based reclamation），在确认没有线程引用该节点后再释放；三是换用 C++20 的std::atomic<std::shared_ptr<T>>（需要 C++20），靠引用计数规避节点被提前释放——C++17 里的替代方案是 C++11 就有的std::atomic_load/std::atomic_store自由函数作用于shared_ptr。

需要把话说透：无锁不等于快，也不等于简单。上面这个pop只是教学骨架——不回收内存（会泄漏）、不处理 ABA，也没解决"何时能安全删除节点"这个真正困难的问题。生产代码里，一个加std::mutex的栈在绝大多数场景下都更短、更对、更容易维护。

常见坑点

再补一条容易被忽略的：不要把std::mutex的地址跨线程传递后再依赖它的生命周期。互斥量本身也是对象，如果它随某个对象一起析构了，而另一个线程还在lock()它，那就是访问已销毁对象，属于 UB。

总结

选同步方式的顺序建议是：先用互斥量把逻辑写对，确认它确实是瓶颈之后再考虑替换。无锁代码要额外承担 ABA、内存回收、伪失败重试和"改一行就出错"的维护成本；真正需要它的场景，是临界区短到加锁开销已占大头、并发度极高的热点路径。在那之前，把锁的粒度缩小、把临界区里的耗时操作（I/O、内存分配、日志）挪出去，往往收益更大也更安全。