C++内存模型

前置知识#

• C++ 面向对象进阶：建议先完成前一篇的学习

学习目标#

• 掌握「历史动机与背景」的核心机制、典型用法与常见陷阱
• 掌握「形式化定义」的核心机制、典型用法与常见陷阱
• 掌握「理论推导」的核心机制、典型用法与常见陷阱
• 掌握「代码示例」的核心机制、典型用法与常见陷阱
• 掌握「对比分析」的核心机制、典型用法与常见陷阱

历史动机与背景#

C++ 内存模型的诞生是并发编程领域数十年的工程实践与理论研究的结晶。

1970-1990:硬件层面的早期探索#

早期单处理器时代,内存模型问题不存在。随着多处理器系统出现,IBM 370、VAX、x86 等架构展现出不同的内存可见性行为。1979 年 Leslie Lamport 发表《How to Make a Multiprocessor Computer That Correctly Executes Multiprocess Programs》,首次形式化定义了”顺序一致性”(Sequential Consistency)概念,奠定现代内存模型的理论基础。

1990-2000:Java 的首次尝试#

1996 年 Java 1.0 引入 volatile 关键字试图解决可见性问题,但语义过于模糊。直到 2004 年 JSR-133 才完成 Java Memory Model (JMM) 的重新设计,引入 happens-before、synchronizes-with 等概念,成为后续语言内存模型的范本。Jeremy Manson、Bill Pugh 等人的工作直接影响了 C++ 标准。

2002-2008:C++ 标准化的艰难历程#

C++98/C++03 没有线程概念,所有多线程行为都依赖平台扩展 (POSIX threads、Windows threads)。2002 年 Boost.Thread 启动,2004 年 Boost.Interatomic 尝试提供跨平台原子操作。Boost 这两项工作成为 C++11 并发标准的原型。

C++ 内存模型的核心设计师 Hans-J. Boehm (HP Labs) 在 2005-2008 年间发表了一系列论文,论证了”无数据竞争”(data-race-free) 语义的必要性。他与 Mark Batty、Peter Sewell (Cambridge) 等人合作,利用形式化方法验证了 C++ 内存模型的一致性。Peter Sewell 团队的 memevents 工具对 ARM、POWER、x86 进行了详尽的 litmus test 测试,揭示了真实硬件与直觉的巨大差距。

2008-2011:C++11 标准确立#

C++11 (ISO/IEC 14882:2011) 第 1.10 节正式定义内存模型,核心要点:

• 引入”数据竞争导致未定义行为”原则 (data-race-free DRF-SC)
• 定义六种内存序,提供从弱到强的灵活选择
• 引入 std::atomic、std::atomic_flag、std::thread、std::mutex 等
• 形式化 happens-before、synchronizes-with、sequenced-before 关系

2011-2020:C++14/17/20 的渐进完善#

C++14 主要是措辞完善;自 C++17 起,memory_order_consume 的规范进入修订状态、标准临时不建议使用(实现上没有任何主流编译器真正追踪依赖链,一律按 acquire 处理;C++26 起正式废弃)。C++20 引入(注意:std::atomic_ref 是 C++20 特性,常被误记为 C++17):

• std::atomic::wait/notify_one/notify_all,提供 futex 风格的高效等待
• std::atomic_ref,允许对非原子对象进行原子访问
• std::counting_semaphore、std::latch、std::barrier 等同步原语
• std::jthread 与协作式取消机制

2020 至今:C++23/26 与硬件演进#

C++23 继续增补原子 API(如 std::atomic 的 fetch_min/fetch_max)。std::execution(Sender/Receiver 模型)已并入 C++26 草案(P2300),为异步并发提供标准执行框架。

硬件层面,ARMv8、RISC-V 等弱内存模型架构在数据中心与移动端普及,x86-TSO 不再是默认假设。Apple Silicon M1/M2、AWS Graviton、华为鲲鹏等 ARM 处理器迫使 C++ 开发者重新审视内存序选择。同时,GPU (CUDA、HIP) 与 CXL (Compute Express Link) 等异构内存架构进一步复杂化了可见性问题。

核心动因总结#

C++ 内存模型的演化动因可归纳为:

• 跨平台可移植性:统一 x86、ARM、POWER、RISC-V 等不同架构的并发语义
• 性能优化:细粒度内存序允许在不牺牲正确性的前提下放松硬件屏障
• 形式化严谨性:利用数学证明保证并发程序正确性,避免”恰好工作”的脆弱代码
• 编译器优化空间:明确定义数据竞争 UB,允许编译器进行激进优化

形式化定义#

内存模型的核心问题#

内存模型回答一个根本问题:在多线程程序中,一个线程对内存位置的写操作,何时以及如何对其他线程可见?形式化地,内存模型定义了:

其中 $\mathcal{O}$ 是所有内存操作集合,$\xrightarrow{hb}$ 是 happens-before 关系,$\xrightarrow{sw}$ 是 synchronizes-with 关系,$\xrightarrow{sb}$ 是 sequenced-before 关系。

Sequenced-Before 关系#

在同一线程内,操作之间的 sequenced-before 关系 $\xrightarrow{sb}$ 由求值规则决定:

例如 int x = 1; int y = x + 1; 中,x=1 sequenced-before y=x+1。但子表达式求值顺序 (f(g(), h())) 在某些情况下未指定,不构成 sequenced-before。

Synchronizes-With 关系#

跨线程的同步通过 synchronizes-with 关系 $\xrightarrow{sw}$ 建立:

具体而言,若线程 T1 执行 x.store(v, release),线程 T2 执行 x.load(acquire) 且读到 v,则 x.store synchronizes-with x.load。

Happens-Before 关系#

happens-before 关系 $\xrightarrow{hb}$ 是 sequenced-before 与 synchronizes-with 的传递闭包:

即 $a \xrightarrow{hb} b$ 当且仅当存在有限序列 $a = o_1, o_2, \ldots, o_n = b$,使得每对相邻 $o_i \xrightarrow{sb} o_{i+1}$ 或 $o_i \xrightarrow{sw} o_{i+1}$。

数据竞争的形式化定义#

给定两个操作 $o_1, o_2$,若满足:

1. $o_1, o_2$ 访问同一内存位置
2. 至少一个是写操作
3. $o_1 \not\xrightarrow{hb} o_2$ 且 $o_2 \not\xrightarrow{hb} o_1$
4. 二者在不同线程

则构成数据竞争 (data race)。C++ 标准规定:无数据竞争的程序在顺序一致性语义下执行 (DRF-SC)。

内存序的数学语义#

六种内存序的行为可形式化为对重排序的限制:

顺序一致性的形式化#

顺序一致性 (Sequential Consistency, SC) 要求存在单一全局总序 $<_\text{sc}$,满足:

1. 每个 seq_cst 操作在 $<_\text{sc}$ 中线性排列
2. $<_\text{sc}$ 与各线程的 sequenced-before 一致
3. $<_\text{sc}$ 与 synchronizes-with 一致
4. 每个 seq_cst 读操作读到最近一次 seq_cst 写操作的值

修改顺序 (Modification Order)#

每个原子对象有一个修改顺序 $<_\text{mo}$,即所有写操作的单一全序。读操作看到的值必须按 $<_\text{mo}$ 单调推进 (coherence):

Cache Coherence 与硬件实现#

硬件层缓存一致性通常采用 MESI 协议,定义四种缓存行状态:

MESI 保证单写多读一致性,但不保证全局顺序。x86-TSO 在 MESI 之上增加 Store Buffer 与 Memory Ordering Machine Clear,提供”Total Store Order”。ARM/RISC-V 则允许更激进的重排,需要显式屏障 (DMB/DSB/ISB)。

理论推导#

Store Buffer 导致的可见性延迟#

x86-TSO 模型中,每个核心有 Store Buffer。考虑:

// 线程 0          // 线程 1
x = 1;             r1 = y;  // 0
y = 1;             r2 = x;  // 0?

在严格 SC 下,r1=0, r2=0 不可能 (因为 x=1 与 y=1 中至少一个先发生)。但在 TSO 下,x=1 进入 Store Buffer,线程 0 继续执行 y=1;线程 1 看到 y=0 后读 x 也可能从自己的缓存读到 0。结果 (r1, r2) = (0, 0) 在 TSO 下合法。

形式化描述:TSO 允许 store-store 重排,但禁止 load-load、load-store、store-load (后者需特殊情况)。即:

Dekker 互斥在弱内存模型下失效#

Dekker 算法经典实现:

// 线程 0                // 线程 1
flag0 = true;            flag1 = true;
if (flag1) {             if (flag0) {
    flag0 = false;           flag1 = false;
    // 等待重试             // 等待重试
} else {                 } else {
    // 临界区               // 临界区
}                        }

在 SC 下,两个线程不能同时进入临界区。但在 ARM 等弱内存模型下,flag0 = true 与 if (flag1) 之间无屏障,可能被重排,导致两者同时看到对方 flag 为 false。修复需要 atomic_thread_fence(seq_cst) 或使用 seq_cst 原子。

ABA 问题的形式化#

考虑无锁栈的 pop 操作:

Node* old = head.load();
Node* next = old->next;
if (head.compare_exchange_weak(old, next)) { /* 成功 */ }

ABA 问题:线程 T1 读取 head = A,被抢占;T2 pop A, pop B, push A (复用节点)。T1 唤醒后 CAS(A, next) 仍成功,但 A->next 可能指向已释放的 B。

形式化:

解决方案:版本号 (tagged pointer)、hazard pointer、epoch-based reclamation。

CAS 复杂度与无锁进度保证#

无锁 (lock-free) 算法的进度保证层次:

CAS 循环通常只能保证 lock-free。若所有线程同时竞争,理论上可能无限重试。Herlihy & Shavit 证明,使用 CAS 实现的任意共识对象存在共识数 (consensus number) 限制:CAS 的共识数为 $\infty$,但读-改-写 (RMW) 操作的共识数有限。

内存序与硬件屏障映射#

不同内存序在不同架构上映射为不同指令:

x86 上 seq_cst store 需要额外 MFENCE,开销显著。ARMv8 的 LDAR/STLR 设计使得 acquire/release 几乎无开销,但 seq_cst 仍需 DMB。

伪共享的性能影响#

缓存行通常 64 字节。若两个原子变量在同一缓存行,跨核修改会触发 MESI 协议的 invalidate 风暴,性能下降 10-100 倍。形式化:

而独立缓存行时:

使用 alignas(64) 或 [[likely]]/[[unlikely]] 属性可缓解。

代码示例#

示例 1:原子计数器与多种内存序对比#

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>
#include <vector>

// 演示不同内存序在原子计数器场景下的正确性与性能差异
class AtomicCounter {
    std::atomic<long> value_{0};
public:
    // relaxed 版本:仅保证原子性,不保证其他变量可见性
    void increment_relaxed() {
        value_.fetch_add(1, std::memory_order_relaxed);
    }

    // seq_cst 版本:最强保证,开销最大
    void increment_seq_cst() {
        value_.fetch_add(1, std::memory_order_seq_cst);
    }

    // acq_rel 版本:RMW 操作的折中选择
    void increment_acq_rel() {
        value_.fetch_add(1, std::memory_order_acq_rel);
    }

    long load() const {
        return value_.load(std::memory_order_acquire);
    }
};

// 性能基准测试
void benchmark(const char* name, void (AtomicCounter::*op)()) {
    AtomicCounter counter;
    constexpr int kThreads = 8;
    constexpr int kIters = 10'000'000;
    std::vector<std::thread> threads;
    auto start = std::chrono::high_resolution_clock::now();
    for (int i = 0; i < kThreads; ++i) {
        threads.emplace_back([&counter, op]() {
            for (int j = 0; j < kIters; ++j) {
                (counter.*op)();
            }
        });
    }
    for (auto& t : threads) t.join();
    auto end = std::chrono::high_resolution_clock::now();
    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << name << ": " << ms << " ms, result=" << counter.load() << "\n";
}

int main() {
    // relaxed 最快,seq_cst 最慢,acq_rel 居中
    benchmark("relaxed ", &AtomicCounter::increment_relaxed);
    benchmark("acq_rel ", &AtomicCounter::increment_acq_rel);
    benchmark("seq_cst ", &AtomicCounter::increment_seq_cst);
    return 0;
}

示例 2:Release-Acquire 模式的生产者-消费者#

#include <atomic>
#include <chrono>
#include <iostream>
#include <string>
#include <thread>

// 演示 release-acquire 配对建立 happens-before 关系
// 生产者写入数据后 release store 标志位,消费者 acquire load 看到标志位后必定看到数据
struct Payload {
    int id;
    double value;
    char name[32];
};

Payload shared_payload;
std::atomic<bool> ready{false};

// 生产者线程:先填充数据,再 release store 通知消费者
void producer() {
    // 1. 准备数据 (sequenced-before store)
    shared_payload.id = 42;
    shared_payload.value = 3.14159;
    std::snprintf(shared_payload.name, sizeof(shared_payload.name), "example");

    // 2. release store:之前的所有写入对 acquire 读取者可见
    //    标准保证:shared_payload 的写入 happens-before ready.store
    ready.store(true, std::memory_order_release);
}

// 消费者线程:acquire load 等待标志位,看到后必定能看到生产者的全部写入
void consumer() {
    // 1. 自旋等待 ready 变为 true
    //    acquire load:之后的读取不会被重排到此 load 之前
    while (!ready.load(std::memory_order_acquire)) {
        std::this_thread::yield();  // 让出 CPU,避免空转消耗
    }

    // 2. 此时 shared_payload 的所有字段必定可见且一致
    //    因为 ready.store (release) synchronizes-with ready.load (acquire)
    //    而 producer 中 payload 填充 sequenced-before store
    //    consumer 中 load sequenced-before payload 读取
    //    根据 happens-before 传递性,数据写入 happens-before 数据读取
    std::cout << "id=" << shared_payload.id
              << " value=" << shared_payload.value
              << " name=" << shared_payload.name << "\n";
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer);
    t1.join();
    t2.join();
    return 0;
}

示例 3:无锁 SPSC 环形队列#

#include <atomic>
#include <optional>
#include <new>

// 单生产者单消费者无锁队列,基于环形缓冲区
// 关键点:
// 1. head 与 tail 分别由生产者与消费者独占修改,避免 CAS 竞争
// 2. 使用 alignas(64) 避免 head 与 tail 落入同一缓存行 (伪共享)
// 3. 使用 acquire-release 配对同步数据访问
template <typename T, size_t Capacity>
class SPSCQueue {
    // 缓存行对齐,避免 head 与 tail 伪共享
    alignas(64) std::atomic<size_t> head_{0};  // 消费者读位置
    alignas(64) std::atomic<size_t> tail_{0};  // 生产者写位置
    alignas(64) T buffer_[Capacity];           // 数据存储

public:
    // 生产者调用:将 item 入队
    bool push(const T& item) {
        const size_t tail = tail_.load(std::memory_order_relaxed);
        const size_t next_tail = (tail + 1) % Capacity;

        // 检查队列是否已满:读取 head (acquire)
        // 同步消费者的 pop 操作:消费者 pop 后 release store head
        if (next_tail == head_.load(std::memory_order_acquire)) {
            return false;  // 队列满
        }

        // 写入数据,然后 release store tail 通知消费者
        buffer_[tail] = item;
        tail_.store(next_tail, std::memory_order_release);
        return true;
    }

    // 消费者调用:出队
    std::optional<T> pop() {
        const size_t head = head_.load(std::memory_order_relaxed);

        // 检查队列是否为空:读取 tail (acquire)
        // 同步生产者的 push 操作
        if (head == tail_.load(std::memory_order_acquire)) {
            return std::nullopt;  // 队列空
        }

        // 读取数据,然后 release store head 通知生产者
        T item = buffer_[head];
        head_.store((head + 1) % Capacity, std::memory_order_release);
        return item;
    }
};

// 使用示例
int main() {
    SPSCQueue<int, 1024> queue;
    std::atomic<bool> done{false};

    // 生产者线程
    std::thread producer([&]() {
        for (int i = 0; i < 100000; ++i) {
            while (!queue.push(i)) {
                std::this_thread::yield();
            }
        }
        done.store(true, std::memory_order_release);
    });

    // 消费者线程
    std::thread consumer([&]() {
        int sum = 0;
        while (true) {
            auto item = queue.pop();
            if (item) {
                sum += *item;
            } else if (done.load(std::memory_order_acquire)) {
                // 队列空且生产者已完成
                if (!queue.pop()) break;
            }
        }
        std::cout << "sum=" << sum << "\n";
    });

    producer.join();
    consumer.join();
    return 0;
}

示例 4:CAS 实现的无锁栈#

#include <atomic>
#include <iostream>
#include <thread>

// 基于 CAS 的无锁栈 (Treiber Stack)
// 注意:此实现存在 ABA 问题,实际生产环境需配合 hazard pointer 或 RCU
template <typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
        Node(const T& d) : data(d), next(nullptr) {}
    };

    std::atomic<Node*> head_{nullptr};

public:
    void push(const T& value) {
        Node* new_node = new Node(value);
        // 1. 读取当前 head (relaxed:不需要同步,仅作为 CAS 期望值)
        new_node->next = head_.load(std::memory_order_relaxed);

        // 2. CAS 循环:尝试将 head 从 new_node->next 改为 new_node
        //    成功用 release:确保 new_node 的写入对 acquire 读取者可见
        //    失败用 relaxed:其他线程的修改不需要同步到本线程
        while (!head_.compare_exchange_weak(
            new_node->next,  // 期望值 (会被自动更新)
            new_node,        // 欲写入值
            std::memory_order_release,
            std::memory_order_relaxed)) {
            // CAS 失败,new_node->next 已被更新为最新 head,继续重试
        }
    }

    // 出栈:返回栈顶元素,空栈返回 false
    bool pop(T& result) {
        Node* old_head = head_.load(std::memory_order_acquire);
        while (old_head) {
            // acquire 读取 old_head 后,old_head->next 必定可见
            // (因为 push 时 release store 保证)
            Node* next = old_head->next;

            // 尝试将 head 从 old_head 改为 next
            if (head_.compare_exchange_weak(
                old_head,
                next,
                std::memory_order_acquire,
                std::memory_order_relaxed)) {
                // 成功:old_head 已被弹出
                result = old_head->data;
                // 注意:此处存在 use-after-free 风险 (ABA 问题)
                // 实际生产环境需使用 hazard pointer 延迟回收
                delete old_head;
                return true;
            }
            // CAS 失败,old_head 已被更新,继续重试
        }
        return false;  // 栈空
    }
};

int main() {
    LockFreeStack<int> stack;
    std::vector<std::thread> threads;

    // 多线程并发 push
    for (int t = 0; t < 4; ++t) {
        threads.emplace_back([&stack, t]() {
            for (int i = 0; i < 1000; ++i) {
                stack.push(t * 1000 + i);
            }
        });
    }

    // 多线程并发 pop
    std::atomic<int> popped_count{0};
    for (int t = 0; t < 4; ++t) {
        threads.emplace_back([&stack, &popped_count]() {
            int value;
            while (stack.pop(value)) {
                popped_count.fetch_add(1, std::memory_order_relaxed);
            }
        });
    }

    for (auto& th : threads) th.join();
    std::cout << "popped " << popped_count.load() << " items\n";
    return 0;
}

示例 5:双重检查锁定 (DCLP) 的正确实现#

#include <atomic>
#include <memory>
#include <mutex>

// 双重检查锁定模式:延迟初始化且避免重复加锁
// 错误实现分析:
//   if (ptr == nullptr) { lock; if (ptr == nullptr) ptr = new Obj; unlock; }
//   问题:ptr = new Obj 分为 (1) 分配内存 (2) 构造对象 (3) 赋值给 ptr
//   编译器可能重排为 (1)(3)(2),其他线程看到 ptr 非空但对象未构造
class Singleton {
    static std::atomic<Singleton*> instance_;
    static std::mutex mtx_;
    Singleton() = default;

public:
    static Singleton* getInstance() {
        // 第一次检查 (acquire):与 release store 配对,确保看到完整对象
        Singleton* tmp = instance_.load(std::memory_order_acquire);
        if (tmp == nullptr) {
            // 加锁,串行化初始化
            std::lock_guard<std::mutex> lock(mtx_);
            // 第二次检查 (relaxed):已在锁内,无需同步
            tmp = instance_.load(std::memory_order_relaxed);
            if (tmp == nullptr) {
                tmp = new Singleton();  // 构造完整对象
                // release store:确保构造的写入对 acquire 读取者可见
                instance_.store(tmp, std::memory_order_release);
            }
        }
        return tmp;
    }
};

std::atomic<Singleton*> Singleton::instance_{nullptr};
std::mutex Singleton::mtx_;

// C++11 后更推荐使用局部静态变量,编译器保证线程安全初始化
class ModernSingleton {
    ModernSingleton() = default;
public:
    static ModernSingleton& getInstance() {
        // C++11 [stmt.dcl]/4:局部静态变量初始化线程安全
        static ModernSingleton instance;
        return instance;
    }
};

int main() {
    auto* p1 = Singleton::getInstance();
    auto* p2 = Singleton::getInstance();
    std::cout << "same instance: " << (p1 == p2) << "\n";
    return 0;
}

示例 6:C++20 wait/notify 高效等待#

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

// C++20 引入 atomic::wait/notify,底层使用 futex (Linux) 或类似机制
// 相比自旋或 condition_variable,大幅降低 CPU 占用
class EventFlag {
    std::atomic<int> flag_{0};
public:
    // 等待方:阻塞直到 flag 不等于 expected
    void wait(int expected) {
        // wait 内部会处理虚假唤醒:每次唤醒后重新检查
        flag_.wait(expected, std::memory_order_acquire);
    }

    // 通知方:修改 flag 并唤醒一个等待线程
    void set_and_notify_one(int value) {
        flag_.store(value, std::memory_order_release);
        flag_.notify_one();  // 唤醒一个等待线程
    }

    // 通知所有等待线程
    void set_and_notify_all(int value) {
        flag_.store(value, std::memory_order_release);
        flag_.notify_all();
    }

    int load() const { return flag_.load(std::memory_order_acquire); }
};

int main() {
    EventFlag event;

    // 等待线程
    std::thread waiter([&]() {
        std::cout << "waiter: waiting for signal...\n";
        event.wait(0);  // 阻塞直到 flag != 0
        std::cout << "waiter: received signal, flag=" << event.load() << "\n";
    });

    // 通知线程
    std::thread notifier([&]() {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        std::cout << "notifier: sending signal\n";
        event.set_and_notify_one(42);
    });

    waiter.join();
    notifier.join();
    return 0;
}

示例 7:内存屏障 fence 的灵活用法#

#include <atomic>
#include <iostream>
#include <thread>

// std::atomic_thread_fence 提供独立的内存屏障
// 与原子操作配合,可实现更灵活的同步模式
// 适用场景:多个数据变量只需一个屏障,而非每个都用 acquire/release

int data1 = 0, data2 = 0, data3 = 0;
std::atomic<bool> flag{false};

// 写入方:先写多个数据,再用单个 fence + relaxed store
void writer() {
    data1 = 100;
    data2 = 200;
    data3 = 300;

    // release fence:之前的所有写入不会被重排到 fence 之后
    std::atomic_thread_fence(std::memory_order_release);

    // relaxed store:仅修改 flag,无额外同步开销
    flag.store(true, std::memory_order_relaxed);
}

// 读取方:relaxed load + acquire fence
void reader() {
    // relaxed load:仅检查 flag
    while (!flag.load(std::memory_order_relaxed)) {
        // 自旋等待
    }

    // acquire fence:之后的所有读取不会被重排到 fence 之前
    std::atomic_thread_fence(std::memory_order_acquire);

    // 此时 data1/2/3 必定可见且一致
    std::cout << "data1=" << data1 << " data2=" << data2 << " data3=" << data3 << "\n";
}

int main() {
    std::thread w(writer);
    std::thread r(reader);
    w.join();
    r.join();
    return 0;
}

示例 8:引用计数智能指针的线程安全实现#

#include <atomic>
#include <iostream>

// 演示 std::shared_ptr 内部引用计数的原子操作原理
// shared_ptr 使用分离的引用计数:强引用 (shared) 与弱引用 (weak)
template <typename T>
class IntrusiveSharedPtr {
    struct ControlBlock {
        std::atomic<long> strong_count{1};
        std::atomic<long> weak_count{1};  // weak_count 包含 strong 引用
        T* ptr;

        ControlBlock(T* p) : ptr(p) {}

        void destroy() {
            delete ptr;
            // 强引用归零,弱引用减一
            if (weak_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
                delete this;
            }
        }
    };

    ControlBlock* control_{nullptr};

public:
    explicit IntrusiveSharedPtr(T* p = nullptr) : control_(p ? new ControlBlock(p) : nullptr) {}

    // 拷贝构造:增加强引用计数
    IntrusiveSharedPtr(const IntrusiveSharedPtr& other) : control_(other.control_) {
        if (control_) {
            // relaxed:仅增加计数,不影响对象本身的可见性
            // 因为其他线程持有时对象必定有效
            control_->strong_count.fetch_add(1, std::memory_order_relaxed);
        }
    }

    // 析构:减少强引用计数
    ~IntrusiveSharedPtr() {
        if (control_) {
            // release:确保对此对象的所有写入在减计数之前完成
            if (control_->strong_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
                // acquire:与之前的 release 配对,确保看到所有写入
                control_->destroy();
            }
        }
    }

    T& operator*() const { return *control_->ptr; }
    T* operator->() const { return control_->ptr; }
    long use_count() const {
        return control_ ? control_->strong_count.load(std::memory_order_relaxed) : 0;
    }
};

int main() {
    IntrusiveSharedPtr<int> p1(new int(42));
    std::cout << "count after p1: " << p1.use_count() << "\n";

    {
        IntrusiveSharedPtr<int> p2 = p1;
        std::cout << "count after p2: " << p1.use_count() << "\n";
    }

    std::cout << "count after p2 destroyed: " << p1.use_count() << "\n";
    std::cout << "value: " << *p1 << "\n";
    return 0;
}

对比分析#

内存序性能对比#

x86 上 acquire/release 几乎免费 (普通 MOV),seq_cst store 需要 MFENCE (约 30-50 周期)。ARMv8 上 LDAR/STLR 比 relaxed 贵约 2-3 周期,seq_cst 需 DMB ISH (约 50-100 周期)。

同步原语对比#

无锁 vs 互斥锁#

跨架构内存序映射#

x86 上 seq_cst store 显著贵于 release,但 ARMv8 上 LDAR/STLR 设计使得 acquire/release 几乎免费。这导致同一段代码在不同架构上性能差异巨大。

atomic vs atomic_ref vs volatile#

volatile 在 C++ 中不提供线程安全,仅用于 MMIO 等场景。Java 的 volatile 等价于 C++ 的 atomic<T> with seq_cst,二者语义完全不同。

常见陷阱与反模式#

反模式 1:误用 relaxed 同步非原子变量#

// 错误:relaxed 不提供任何同步语义
int data = 0;
std::atomic<bool> ready{false};

// 线程 A
data = 42;
ready.store(true, std::memory_order_relaxed);  // data 对其他线程可能不可见!

// 线程 B
if (ready.load(std::memory_order_relaxed)) {
    std::cout << data;  // 未定义行为:可能读到 0
}

生产事故:某高频交易系统使用 relaxed 同步订单簿,在 ARM 服务器上出现订单丢失,定位耗时 3 周。修复改为 release-acquire 后问题消失。

正确做法:

data = 42;
ready.store(true, std::memory_order_release);  // 确保 data 可见

if (ready.load(std::memory_order_acquire)) {
    std::cout << data;  // 安全:看到 42
}

反模式 2:误用 volatile 实现线程安全#

// 错误:C++ 的 volatile 不提供原子性或可见性
volatile int counter = 0;
void increment() {
    counter++;  // 非原子,多线程下数据竞争
}

原因:C++ volatile 仅阻止编译器优化 (如寄存器缓存),不生成内存屏障,不保证 CPU 层原子性。这是 C++ 与 Java 的关键差异。

反模式 3:DCLP 错误实现#

// 错误:双重检查锁定未使用原子
Singleton* Singleton::getInstance() {
    if (instance_ == nullptr) {  // 第一次检查:数据竞争
        std::lock_guard<std::mutex> lock(mtx_);
        if (instance_ == nullptr) {
            instance_ = new Singleton();  // 重排风险
        }
    }
    return instance_;
}

问题:new Singleton() 分三步:(1) 分配 (2) 构造 (3) 赋值。编译器可能重排为 (1)(3)(2),其他线程看到非空 instance_ 但对象未构造完成。

反模式 4:无锁代码的 ABA 问题#

// 错误:CAS 无法检测 ABA 切换
Node* old = head.load();
while (!head.compare_exchange_weak(old, old->next)) {}
// 若 head: A -> B -> C,线程 T1 读到 old=A
// T2 pop A, pop B, push A (复用),T1 CAS 成功但 A->next 可能指向已释放的 B

生产事故:某数据库系统使用 Treiber Stack 管理内存块,在高负载下出现段错误。根因是 ABA 导致访问已释放内存。修复使用 tagged pointer (附加版本号) 后稳定。

反模式 5:伪共享导致性能骤降#

// 错误:多个原子变量在同一缓存行
struct Counters {
    std::atomic<int> a{0};
    std::atomic<int> b{0};
    std::atomic<int> c{0};
};
// 多线程分别修改 a/b/c,触发 MESI invalidate 风暴

性能影响:某并行计算程序,8 线程计数器从 64M ops/s 降至 1.2M ops/s。修复使用 alignas(64) 后恢复至 62M ops/s。

反模式 6:compare_exchange 强弱误用#

// 错误:在循环中使用 strong (多余的开销)
while (!head.compare_exchange_strong(old, new_node)) {}

// 正确:循环中用 weak,避免重复的强保证开销
while (!head.compare_exchange_weak(old, new_node)) {}

原因:weak 允许虚假失败 (spurious failure),即 CAS 本可成功但返回 false。循环中重试即可,而 strong 内部循环直到非虚假结果,在循环中调用是浪费。

反模式 7:seq_cst 滥用导致性能下降#

// 错误:对性能敏感的计数器使用 seq_cst
std::atomic<long> counter{0};
counter.fetch_add(1, std::memory_order_seq_cst);  // 每次 MFENCE,慢 5-10 倍

// 正确:计数器用 relaxed 即可
counter.fetch_add(1, std::memory_order_relaxed);

反模式 8:memory_order_consume 的不可靠性#

// 不推荐:consume 在主流编译器上被当作 acquire 处理,且标准建议避免使用
atomic<T*> ptr;
T* p = ptr.load(std::memory_order_consume);
// 编译器无法正确追踪依赖链,实际效果不确定

// 推荐:直接使用 acquire
T* p = ptr.load(std::memory_order_acquire);

反模式 9:假设原子操作是 wait-free#

// 错误:假设所有原子操作都无锁
std::atomic<BigStruct> big_atomic;  // 大于机器字长
big_atomic.load();  // 可能内部使用 mutex,非 lock-free

// 检查:使用 is_lock_free 或 is_always_lock_free
static_assert(decltype(big_atomic)::is_always_lock_free);

反模式 10:跨线程调用未同步的 this 捕获#

// 错误:Lambda 捕获 this,跨线程访问成员
class Worker {
    int data_{0};
public:
    void start() {
        std::thread([this]() {
            data_ = 42;  // 与主线程的 data_ 访问未同步,数据竞争
        }).detach();
    }
};

工程实践#

实践 1:优先使用标准库而非手写无锁代码#

// 推荐:使用 std::shared_ptr, std::mutex 等经过验证的实现
auto ptr = std::make_shared<BigData>();
std::lock_guard<std::mutex> lock(mtx_);

// 不推荐:手写无锁引用计数 (极易出错)
// 除非有明确的性能瓶颈且已 profile

实践 2:使用 ThreadSanitizer 检测数据竞争#

# 编译时加 -fsanitize=thread
g++ -fsanitize=thread -g -O1 main.cpp -o main
./main  # TSan 会在数据竞争时报告

TSan 能检测 95%+ 的数据竞争,开销约 5-10 倍。CI 流水线中应集成 TSan 测试。

实践 3:使用 LITMUS 测试验证内存序理解#

// 经典 MP (Message Passing) litmus test
std::atomic<int> x{0}, y{0};

void thread0() {
    x.store(1, std::memory_order_relaxed);
    y.store(1, std::memory_order_release);
}

void thread1(int& r1, int& r2) {
    r1 = y.load(std::memory_order_acquire);
    r2 = x.load(std::memory_order_relaxed);
    // 不可能出现 r1=1, r2=0 (release-acquire 保证)
}

实践 4:合理选择数据布局避免伪共享#

// 缓存行对齐的多线程计数器
struct alignas(64) AlignedCounter {
    std::atomic<long> value{0};
};

struct Counters {
    AlignedCounter counters[8];  // 每个计数器独占缓存行
};

// 或使用 padding
struct PaddedCounter {
    std::atomic<long> value{0};
    char padding[64 - sizeof(std::atomic<long>)];
};

实践 5:无锁代码的内存回收#

// Hazard Pointer:延迟回收,避免 use-after-free
template <typename T>
class HazardPointer {
    std::atomic<T*> ptr_{nullptr};
public:
    T* protect(std::atomic<T*>& src) {
        T* p;
        do {
            p = src.load(std::memory_order_acquire);
            ptr_.store(p, std::memory_order_release);
        } while (p != src.load(std::memory_order_acquire));
        return p;
    }

    void release() { ptr_.store(nullptr, std::memory_order_release); }
};

// 全局 hazard pointer 列表,回收时检查
void retire(T* p) {
    // 若 p 不在任何 hazard pointer 中,安全删除
    // 否则加入待回收列表,稍后再试
}

实践 6:性能关键路径的内存序选择#

// 性能关键路径的渐进式优化:
// 1. 先用 seq_cst 保证正确性
// 2. Profile 定位瓶颈
// 3. 仅对瓶颈降级为 acquire/release
// 4. 极端场景才考虑 relaxed

// 示例:高频交易订单匹配引擎
class OrderBook {
    std::atomic<Order*> best_bid_{nullptr};  // seq_cst 保证全局一致

    // 热点路径:用 acq_rel (足够)
    bool try_match(Order* new_order) {
        Order* bid = best_bid_.load(std::memory_order_acquire);
        while (bid && !is_match(bid, new_order)) {
            bid = bid->next;
        }
        if (bid) {
            return best_bid_.compare_exchange_weak(
                bid, bid->next,
                std::memory_order_acq_rel,
                std::memory_order_relaxed);
        }
        return false;
    }
};

实践 7:跨平台可移植性考虑#

// 不可移植:依赖 x86 强内存模型
std::atomic<bool> flag{false};
int data = 0;

// 线程 A
data = 42;
flag.store(true, std::memory_order_relaxed);  // x86 上可能工作,ARM 上失败

// 可移植版本
data = 42;
flag.store(true, std::memory_order_release);  // 所有架构正确

实践 8:使用 C++20 的 atomic_ref 操作非原子对象#

// C++20:原子引用,允许对已有非原子变量进行原子访问
void increment_array(int* arr, size_t n) {
    // arr 是普通 int*,但需要原子操作
    for (size_t i = 0; i < n; ++i) {
        std::atomic_ref<int> ref(arr[i]);
        ref.fetch_add(1, std::memory_order_relaxed);
    }
}

// 适用场景:第三方库返回非原子数据,需原子访问

案例研究#

案例 1:Linux 内核 RCU (Read-Copy-Update)#

Linux RCU 是无锁读取的极致实践。核心思想:

1. 读者无锁:直接访问数据,不加任何锁或原子操作
2. 写者复制:修改时复制一份,修改副本,原子替换指针
3. 延迟回收:等待所有读者退出后,才释放旧数据

// RCU 简化示意
template <typename T>
class RcuProtected {
    std::atomic<T*> ptr_;
public:
    // 读者:无锁访问
    const T* read() {
        return ptr_.load(std::memory_order_consume);  // 依赖顺序
    }

    // 写者:复制-修改-替换
    void update(const T& new_value) {
        T* new_data = new T(new_value);
        T* old = ptr_.exchange(new_data, std::memory_order_acq_rel);
        // synchronize_rcu():等待所有读者完成
        sched_rcu_grace_period();
        delete old;
    }
};

RCU 在 Linux 内核中用于 dcache、page table 等高频路径,读吞吐比 rwlock 高 10-100 倍。Paul McKenney 因此获 2020 Linux 基金会终身成就奖。

案例 2:Facebook Folly 的 MicroLock#

Folly 库针对短临界区设计了 MicroLock,核心优化:

1. 将多个锁压缩到一个 uint64_t 中 (每个锁 8 位)
2. 使用 CAS 而非系统调用
3. 自旋一段时间后回退到 futex

// Folly MicroLock 简化版
class MicroLock {
    std::atomic<uint64_t> state_{0};
    static constexpr uint64_t kLockedBit = 1ULL << 32;
    static constexpr uint64_t kWaiterBit = 1ULL << 33;

public:
    void lock() {
        // 快速路径:CAS 直接获取
        uint64_t expected = state_.load(std::memory_order_relaxed) & ~(kLockedBit | kWaiterBit);
        if (state_.compare_exchange_weak(expected, expected | kLockedBit,
                                          std::memory_order_acquire)) {
            return;  // 成功
        }
        // 慢速路径:自旋 + futex
        slow_lock();
    }
};

MicroLock 在 4 核机器上比 std::mutex 快 5-10 倍,但仅适合极短临界区 (<1μs)。

案例 3:Boost.Lockfree 的队列实现#

Boost.Lockfree 提供生产级无锁队列,关键设计:

1. MPSC 用原子索引:多生产者通过 CAS 推进 tail
2. 内存回收:内部使用 freelist,避免 ABA
3. 可变大小:支持固定大小与动态扩容

#include <boost/lockfree/queue.hpp>

boost::lockfree::queue<int, boost::lockfree::capacity<1024>> q;

// 多生产者多消费者安全
void producer() {
    for (int i = 0; i < 1000; ++i) {
        while (!q.push(i)) {
            std::this_thread::yield();
        }
    }
}

void consumer() {
    int value;
    while (q.pop(value)) {
        process(value);
    }
}

案例 4:ClickHouse 的列式存储并发#

ClickHouse 使用细粒度原子操作实现高并发查询:

1. Part 引用计数:std::shared_ptr 不可见,改用侵入式 atomic 计数
2. Merge 调度:用 atomic flag 避免重复触发
3. 查询取消:原子 bool 定期检查

// ClickHouse 的 MergeTask 简化
class MergeTask {
    std::atomic<bool> cancelled_{false};

    void execute() {
        while (!cancelled_.load(std::memory_order_relaxed)) {
            process_batch();
        }
    }

    void cancel() {
        cancelled_.store(true, std::memory_order_relaxed);
    }
};

案例 5:ARM 服务器迁移的内存序陷阱#

某互联网公司从 Intel Xeon 迁移到 AWS Graviton (ARM),发现以下问题:

1. x86 上的”巧合”代码失效:依赖 TSO 的强排序,未加 acquire/release
2. 性能不升反降:seq_cst 在 ARM 上比 x86 贵 10 倍
3. 缓存行大小差异:ARM 系统缓存行 128 字节 (部分),alignas(64) 不足

修复方案:

// 检测缓存行大小
constexpr size_t kCacheLineSize = 
#ifdef __x86_64__
    64;
#elif defined(__aarch64__)
    128;  // 部分 ARM 服务器
#else
    64;
#endif

struct alignas(kCacheLineSize) Counter {
    std::atomic<long> value{0};
};

案例 6:Rust 与 C++ 内存模型对比#

Rust 借鉴 C++ 内存模型但更严格:

Rust 的 Send/Sync trait 在编译期阻止数据竞争,而 C++ 依赖运行时工具 (TSan)。但 C++ 的灵活性允许更激进的优化。

基础题#

1. 以下代码是否存在数据竞争?若存在,如何修复?

int counter = 0;
std::mutex mtx;

void increment() {
    mtx.lock();
    counter++;
    mtx.unlock();
}

int read() {
    return counter;  // 未加锁
}

参考答案要点:存在数据竞争,read 与 increment 并发访问 counter 且无同步。修复:将 counter 改为 std::atomic<int>,或在 read 中加锁。

2. 解释 relaxed、acquire、release 三种内存序的区别,并各举一个适用场景。

参考答案要点:

• relaxed:仅保证原子性,无同步。适用:计数器统计。
• acquire:load 后的读写不重排到 load 前。适用:读取标志位后访问共享数据。
• release:store 前的读写不重排到 store 后。适用:写入数据后设置 ready 标志。

3. 为什么 std::atomic<bool> 通常比 std::mutex 更快?在什么情况下反而更慢?

参考答案要点:atomic 用户态完成,无系统调用开销。但当自旋时间超过上下文切换开销时,mutex 让出 CPU 更高效。

进阶题#

4. 以下代码在 x86 和 ARM 上行为是否一致?为什么?

std::atomic<int> x{0}, y{0};
int r1, r2;

void t1() {
    x.store(1, std::memory_order_relaxed);
    y.store(1, std::memory_order_relaxed);
}

void t2() {
    r1 = y.load(std::memory_order_relaxed);
    r2 = x.load(std::memory_order_relaxed);
}
// 可能出现 r1=1, r2=0 吗?

参考答案要点:

• x86 (TSO):禁止 store-store 重排,r1=1, r2=0 不可能
• ARM (弱):允许 store-store 重排,r1=1, r2=0 可能
• 这体现了跨架构可移植性的关键差异

5. 实现 Peterson 互斥算法,并分析为何在 C++ 默认内存模型下需要 seq_cst。

// Peterson 算法 (简化)
std::atomic<bool> flag[2]{false, false};
std::atomic<int> turn{0};

void lock(int id) {
    int other = 1 - id;
    flag[id].store(true, ???);
    turn.store(other, ???);
    while (flag[other].load(???) && turn.load(???) == other) {}
}

参考答案要点:Peterson 依赖严格的操作顺序,弱内存序下 flag[id]=true 可能被重排到 turn.store 之后,导致两个线程同时进入临界区。必须用 seq_cst 或显式 fence。

6. 分析以下代码的 ABA 问题,并给出两种解决方案。

std::atomic<Node*> head;
Node* old = head.load();
while (old && !head.compare_exchange_weak(old, old->next)) {}

参考答案要点:

• ABA:head 从 A 变 B 再变 A,CAS 成功但 A->next 可能指向已释放的 B
• 方案 1:tagged pointer,附加版本号 atomic<pair<Node*, uint64_t>>
• 方案 2:hazard pointer,读取前登记,回收前检查

挑战题#

7. 设计一个无锁 MPMC 队列,要求:

• 支持多生产者多消费者
• 无 ABA 问题
• 无内存泄漏
• 给出完整 C++20 实现

参考答案要点:基于环形缓冲区 + 序号 (sequence number) 的 Michael-Scott 队列变体,使用 atomic<uint64_t> 编码位置与版本。关键点:

template <typename T, size_t Capacity>
class MPMCQueue {
    struct Cell {
        std::atomic<size_t> sequence;
        T data;
    };
    alignas(64) Cell buffer_[Capacity];
    alignas(64) std::atomic<size_t> enqueue_pos_{0};
    alignas(64) std::atomic<size_t> dequeue_pos_{0};

public:
    bool push(const T& value) {
        Cell* cell;
        size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
        for (;;) {
            cell = &buffer_[pos % Capacity];
            size_t seq = cell->sequence.load(std::memory_order_acquire);
            intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
            if (diff == 0) {
                if (enqueue_pos_.compare_exchange_weak(pos, pos + 1,
                                                       std::memory_order_relaxed)) break;
            } else if (diff < 0) {
                return false;  // 满
            } else {
                pos = enqueue_pos_.load(std::memory_order_relaxed);
            }
        }
        cell->data = value;
        cell->sequence.store(pos + 1, std::memory_order_release);
        return true;
    }
    // pop 类似
};

8. 证明:x86-TSO 上,acquire/release 可以用普通 MOV 实现,但 seq_cst store 必须加 MFENCE。

参考答案要点:

• TSO 已保证 load-load、load-store、store-store 不重排,因此 acquire load (普通 MOV) 后的读写自然在 load 后
• TSO 允许 store-load 重排 (store buffer),因此 seq_cst store 后的 seq_cst load 必须用 MFENCE 刷新 store buffer,否则全局总序无法保证
• 反例:Store Buffer Litmus Test (SB) 在 TSO 下允许 (r1,r2)=(0,0),但 seq_cst 禁止

9. 分析 Linux READ_ONCE/WRITE_ONCE 宏与 std::atomic 的关系,并说明为何内核不使用 C++ 原子。

参考答案要点:

• READ_ONCE/WRITE_ONCE 等价于 relaxed 原子,防止编译器拆分/合并访问
• 内核用 C 实现,无 C++ 编译器;且需要与汇编、volatile MMIO 交互
• 内核内存模型 (LKMM) 比 C++ 更复杂,涉及 RCU、barrier() 等扩展

10. 设计一个测试框架,自动检测并发代码在不同内存序下的正确性。

参考答案要点:

• 使用 LITMUS 测试范式:枚举线程组合,运行 N 次,统计结果分布
• 集成 ThreadSanitizer 检测数据竞争
• 跨架构测试:x86 (基线) vs ARM (弱模型)
• 工具:llvm-lit、concurrency-kit、relacy

官方文档#

• C++ Reference: Atomic operations library (https://en.cppreference.com/w/cpp/atomic)
• C++ Reference: Memory order (https://en.cppreference.com/w/cpp/atomic/memory_order)
• ISO C++ N2153: A simple and efficient memory model for C++ (Boehm 提案)
• ISO C++ P0019: atomic_ref (C++20 提案)
• ISO C++ P2165: atomic wait/notify (C++20 提案)

经典教材#

• Anthony Williams. C++ Concurrency in Action (2nd Edition). Manning, 2019.
• Maurice Herlihy, Nir Shavit. The Art of Multiprocessor Programming (2nd Edition). Morgan Kaufmann, 2012.
• Paul E. McKenney. Is Parallel Programming Hard, And, If So, What Can You Do About It? kernel.org.
• Jeff Preshing. Sane C++ Memory Model 系列博客 (https://preshing.com)

前沿论文#

• Batty et al. “Mathematizing C++ Concurrency” (POPL 2011) - 形式化语义
• Lahav et al. “Repairing Sequential Consistency in C/C++11” (PLDI 2017) - 修复 seq_cst 漏洞
• Chakraborty & Vafeiadis “Validating Optimizations of Concurrent C/C++ Programs” (CGO 2019) - 优化正确性

相关课程#

• MIT 6.006/6.172: Performance Engineering of Software Systems
• Stanford CS149: Parallel Computing
• CMU 15-440: Distributed Systems
• Berkeley CS262a: Advanced Topics in Computer Systems

内存顺序基础#

基本写法：顺序一致顺序 std::memory_order_seq_cst

// 默认最严格的全局一致顺序
a.store(1, std::memory_order_seq_cst);

基本写法：获取释放顺序 std::memory_order_acquire / std::memory_order_release

// 配对使用建立同步关系
data.store(42, std::memory_order_release);
int v = data.load(std::memory_order_acquire);

基本写法：宽松顺序 std::memory_order_relaxed

// 仅保证原子操作本身无竞争
counter.fetch_add(1, std::memory_order_relaxed);

基本写法：消费顺序 std::memory_order_consume

// 依赖数据的获取顺序通常建议用 acquire 替代
int v = data.load(std::memory_order_consume);

同步关系#

基本写法：发布订阅模式 <写线程>.store(<值>, release); <读线程>.load(acquire);

// 线程间建立先行关系
// 线程 A
data = 42;
flag.store(true, std::memory_order_release);
// 线程 B
while (!flag.load(std::memory_order_acquire));
// 此处能看到 data == 42

基本写法：栅栏建立同步 std::atomic_thread_fence(<内存序>);

// 显式屏障替代操作内存序
data = 42;
std::atomic_thread_fence(std::memory_order_release);
flag.store(true, std::memory_order_relaxed);

happens-before 关系#

基本写法：序列先行 <语句1>; <语句2>;

// 同一线程内前者 sequenced-before 后者
int a = 1;
int b = a + 1;

基本写法：同步建立先行 store(release) ↔ load(acquire)

// release 操作 happens-before 配对的 acquire
ready.store(true, std::memory_order_release);

数据竞争#

基本写法：避免数据竞争 std::atomic<<类型>> <变量>;

// 多线程访问共享变量需原子或加锁
std::atomic<int> counter{0};

基本写法：mutex 保护共享数据 std::lock_guard<<锁类型>> <变量>(<锁>);

// 通过锁保证互斥访问
std::mutex m;
std::lock_guard<std::mutex> lk(m);
data.push_back(x);

可见性与顺序#

基本写法：一次性写入 std::call_once(<flag>, <函数>);

// 保证初始化只执行一次
std::once_flag flag;
std::call_once(flag, []{ obj = new Object; });

基本写法：局部静态线程安全 static <类型> <变量>(<参数>);

// C++11 起局部静态初始化线程安全
static Widget& instance() {
    static Widget w;
    return w;
}

volatile 关键字#

基本写法：禁止优化 volatile <类型> <变量>;

// 阻止编译器优化读写用于硬件寄存器
volatile int* reg = (volatile int*)0x4000;

基本写法：注意 volatile 不保证原子性 volatile <类型> <变量>; // 多线程下不安全

// volatile 仅禁优化不提供原子或顺序保证
volatile int v = 0;  // 多线程读写仍需锁或原子

顺序一致性与性能#

基本写法：选择适当内存序 load(acquire) / store(release)

// 性能优于 seq_cst 的常用配对
flag.load(std::memory_order_acquire);
flag.store(true, std::memory_order_release);

基本写法：统计计数用宽松 fetch_add(<值>, relaxed)

// 无顺序要求场景使用最宽松顺序
counter.fetch_add(1, std::memory_order_relaxed);

信号量 C++20#

基本写法：创建信号量 std::counting_semaphore<<最大值>> <变量>(<初始值>);

// 计数信号量
std::counting_semaphore<10> sem(3);

基本写法：获取与释放 <sem>.acquire(); <sem>.release();

// P 操作与 V 操作
sem.acquire();   // 计数减 1
// 临界区
sem.release();   // 计数加 1

基本写法：二元信号量 std::binary_semaphore <变量>(<初始值>);

// 等价于只有 0 和 1 的信号量
std::binary_semaphore sem(1);