C++ 内存序深度解析：6 种内存模型与多线程同步

2. 内存模型基础概念

2.1 原子操作

C++11 提供了 std::atomic 模板，支持原子操作：

#include <atomic>
std::atomic<int> counter{0};

// 线程安全的自增
void increment() {
    counter.fetch_add(1, std::memory_order_relaxed);
}

2.2 内存顺序的类型

C++ 定义了 6 种内存顺序，分为 3 大类：

1. 顺序一致性：memory_order_seq_cst
2. 获取 - 释放语义：memory_order_acquire、memory_order_release、memory_order_acq_rel
3. 宽松顺序：memory_order_relaxed、memory_order_consume（已弃用）

3. 6 种内存序详解

3.1 memory_order_seq_cst（顺序一致性）

最强的一致性保证，是原子操作的默认内存序。

std::atomic<bool> x{false}, y{false};
std::atomic<int> z{0};

void write_x() {
    x.store(true, std::memory_order_seq_cst); // #1
}

void write_y() {
    y.store(true, std::memory_order_seq_cst); // #2
}

void read_x_then_y() {
    while (!x.load(std::memory_order_seq_cst)) {} // #3
    if (y.load(std::memory_order_seq_cst)) { // #4
        ++z;
    }
}

void read_y_then_x() {
    while (!y.load(std::memory_order_seq_cst)) {} // #5
    if (x.load(std::memory_order_seq_cst)) { // #6
        ++z;
    }
}

顺序一致性保证：如果 read_x_then_y 看到 x=true 然后 y=false，那么 read_y_then_x 不可能看到 y=true 然后 x=false。

特点：

• 所有操作按某个全局顺序执行
• 所有线程看到相同的操作顺序
• 性能开销最大

3.2 memory_order_acquire（获取语义）

建立读操作的同步关系，确保后续的读/写操作不会被重排到该操作之前。

std::atomic<bool> ready{false};
int data = 0;

void producer() {
    data = 42; // #1
    ready.store(true, std::memory_order_release); // #2
}

void consumer() {
    while (!ready.load(std::memory_order_acquire)) {} // #3
    std::cout << data << std::endl; // 保证看到 42 // #4
}

获取 - 释放配对保证：

• #1（非原子写）happens-before #2（原子释放存储）
• #3（原子获取加载）synchronizes-with #2
• #4（非原子读）happens-after #3

因此 #1 happens-before #4，消费者保证看到 data = 42。

3.3 memory_order_release（释放语义）

建立写操作的同步关系，确保之前的读/写操作不会被重排到该操作之后。

// 自旋锁的实现
class SpinLock {
    std::atomic<bool> locked{false};
public:
    void lock() {
        while (locked.exchange(true, std::memory_order_acquire)) { // 自旋等待
        }
    }
    void unlock() {
        locked.store(false, std::memory_order_release);
    }
};

关键作用：

• 释放锁时使用 release 语义
• 确保临界区内的所有操作在锁释放前完成
• 让其他线程在获取锁时能看到这些操作

3.4 memory_order_acq_rel（获取 - 释放语义）

同时具有获取和释放语义，用于读 - 修改 - 写操作。

std::atomic<int> counter{0};

void increment() {
    // fetch_add 是读 - 修改 - 写操作
    counter.fetch_add(1, std::memory_order_acq_rel);
}

// 等价于：
void increment_manual() {
    int old = counter.load(std::memory_order_acquire);
    counter.store(old + 1, std::memory_order_release);
}

使用场景：

• 自旋锁的加锁操作（exchange）
• 引用计数的增减
• 任何需要同时建立前后同步关系的操作

3.5 memory_order_relaxed（宽松语义）

最弱的内存序保证，只保证原子性，不保证同步和顺序。

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

void thread1() {
    x.store(1, std::memory_order_relaxed); // #1
    y.store(1, std::memory_order_relaxed); // #2
    // #1 和 #2 可能被重排
}

void thread2() {
    int r1 = y.load(std::memory_order_relaxed); // #3
    int r2 = x.load(std::memory_order_relaxed); // #4
    // 可能看到 r1=1, r2=0
    std::cout << "r1=" << r1 << ", r2=" << r2 << std::endl;
}

适用场景：

• 统计计数器
• 标志位（不需要与其他数据同步）
• 性能要求极高的场景

3.6 memory_order_consume（消费语义，已弃用）

C++17 已弃用，不建议使用。原本用于数据依赖关系的优化。

// 不推荐使用
std::atomic<Data*> data_ptr{nullptr};
int value = 0;

void producer() {
    Data* p = new Data{42};
    value = 100;
    data_ptr.store(p, std::memory_order_release);
}

void consumer() {
    Data* p;
    while (!(p = data_ptr.load(std::memory_order_consume))) {}
    // 保证看到 *p 的正确值，但不保证看到 value=100
    std::cout << p->value << std::endl;
}

4. 代码示例与分析

4.1 双重检查锁定模式

// 线程安全的单例模式
class Singleton {
private:
    static std::atomic<Singleton*> instance;
    static std::mutex mutex;
    Singleton() = default;

public:
    static Singleton* getInstance() {
        Singleton* tmp = instance.load(std::memory_order_acquire); // 第一次检查（无锁）
        if (tmp == nullptr) {
            std::lock_guard<std::mutex> lock(mutex);
            tmp = instance.load(std::memory_order_relaxed); // 第二次检查（持有锁）
            if (tmp == nullptr) {
                tmp = new Singleton(); // 关键：使用 release 确保构造函数完成后再发布指针
                instance.store(tmp, std::memory_order_release);
            }
        }
        return tmp;
    }

    // 禁止拷贝
    Singleton(const Singleton&) = delete;
    Singleton& operator=(const Singleton&) = delete;
};

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

4.2 无锁队列实现

template<typename T>
class LockFreeQueue {
private:
    struct Node {
        T data;
        std::atomic<Node*> next;
        Node(const T& value) : data(value), next(nullptr) {}
    };
    std::atomic<Node*> head;
    std::atomic<Node*> tail;

public:
    LockFreeQueue() {
        Node* dummy = new Node(T());
        head.store(dummy, std::memory_order_relaxed);
        tail.store(dummy, std::memory_order_relaxed);
    }

    ~LockFreeQueue() {
        while (Node* node = head.load(std::memory_order_relaxed)) {
            head.store(node->next.load(std::memory_order_relaxed), std::memory_order_relaxed);
            delete node;
        }
    }

    void enqueue(const T& value) {
        Node* new_node = new Node(value);
        Node* old_tail = nullptr;
        while (true) {
            old_tail = tail.load(std::memory_order_acquire);
            Node* next = old_tail->next.load(std::memory_order_acquire); // 检查 tail 是否仍然正确
            if (old_tail == tail.load(std::memory_order_relaxed)) {
                if (next == nullptr) {
                    // 尝试链接新节点
                    if (old_tail->next.compare_exchange_weak(next, new_node, std::memory_order_release, std::memory_order_relaxed)) {
                        break; // 成功
                    }
                } else {
                    // 帮助推进 tail
                    tail.compare_exchange_weak(old_tail, next, std::memory_order_release, std::memory_order_relaxed);
                }
            }
        }
        // 推进 tail
        tail.compare_exchange_weak(old_tail, new_node, std::memory_order_release, std::memory_order_relaxed);
    }

    bool dequeue(T& value) {
        while (true) {
            Node* old_head = head.load(std::memory_order_acquire);
            Node* old_tail = tail.load(std::memory_order_acquire);
            Node* next = old_head->next.load(std::memory_order_acquire); // 检查一致性
            if (old_head == head.load(std::memory_order_relaxed)) {
                if (old_head == old_tail) {
                    if (next == nullptr) {
                        return false; // 队列为空
                    }
                    // 帮助推进 tail
                    tail.compare_exchange_weak(old_tail, next, std::memory_order_release, std::memory_order_relaxed);
                } else {
                    value = next->data;
                    if (head.compare_exchange_weak(old_head, next, std::memory_order_release, std::memory_order_relaxed)) {
                        delete old_head;
                        return true;
                    }
                }
            }
        }
    }
};

5. 内存序的选择策略

5.1 决策流程图

• 是否读操作写操作读 - 修改 - 写
• 是 / 简单优先否 / 性能优先
• 开始选择内存序需要同步非原子数据？
• 读操作还是写操作？
• 使用 memory_order_relaxed
• 使用 memory_order_acquire
• 使用 memory_order_release
• 使用 memory_order_acq_rel
• 需要全局顺序一致性？
• 使用 memory_order_seq_cst
• 保持原选择验证正确性测试和基准测试
• 结束

5.2 选择指南表格

6. 实际应用场景

6.1 性能计数器

// 高性能统计计数器
class PerformanceCounter {
    // 每个线程有自己的缓存
    struct alignas(64) ThreadLocal {
        // 缓存行对齐，避免伪共享
        int64_t local_count{0};
        char padding[64 - sizeof(int64_t)]; // 填充到缓存行大小
    };
    std::vector<ThreadLocal> thread_local_counts;
    std::atomic<int64_t> global_count{0};

public:
    PerformanceCounter(size_t num_threads) : thread_local_counts(num_threads) {}

    void increment(size_t thread_id) {
        // 线程本地累加，无锁
        thread_local_counts[thread_id].local_count++;
        // 定期同步到全局计数器
        if (thread_local_counts[thread_id].local_count % 1000 == 0) {
            int64_t local = thread_local_counts[thread_id].local_count;
            thread_local_counts[thread_id].local_count = 0;
            // 宽松内存序即可，因为这是独立的统计
            global_count.fetch_add(local, std::memory_order_relaxed);
        }
    }

    int64_t get_total() const {
        // 需要获取所有线程本地的值
        int64_t total = global_count.load(std::memory_order_relaxed);
        for (const auto& local : thread_local_counts) {
            total += local.local_count;
        }
        return total;
    }
};

6.2 发布 - 订阅模式

// 无锁的发布 - 订阅机制
template<typename T>
class Publisher {
    struct SharedData {
        T data;
        std::atomic<int> version{0};
    };
    alignas(64) SharedData shared[2]; // 双缓冲
    std::atomic<int> current_index{0};

public:
    void publish(const T& new_data) {
        int idx = current_index.load(std::memory_order_relaxed);
        int next_idx = 1 - idx; // 切换到另一个缓冲区
        // 准备新数据
        shared[next_idx].data = new_data;
        // 发布：先增加版本号，再切换缓冲区
        shared[next_idx].version.store(shared[idx].version.load(std::memory_order_acquire) + 1, std::memory_order_release);
        // 切换当前缓冲区（release 确保前面的写对其他线程可见）
        current_index.store(next_idx, std::memory_order_release);
    }

    bool subscribe(T& out_data, int& last_version) {
        int idx = current_index.load(std::memory_order_acquire);
        int version = shared[idx].version.load(std::memory_order_acquire);
        if (version != last_version) {
            // 获取数据（acquire 确保看到发布者的所有写）
            out_data = shared[idx].data;
            last_version = version;
            return true;
        }
        return false;
    }
};

7. 性能对比与基准测试

#include <benchmark/benchmark.h>
#include <atomic>
#include <thread>
#include <vector>

// 基准测试：不同内存序的性能对比
static void BM_AtomicIncrement_Relaxed(benchmark::State& state) {
    std::atomic<int> counter{0};
    for (auto _ : state) {
        counter.fetch_add(1, std::memory_order_relaxed);
        benchmark::DoNotOptimize(counter);
    }
}
BENCHMARK(BM_AtomicIncrement_Relaxed);

static void BM_AtomicIncrement_AcqRel(benchmark::State& state) {
    std::atomic<int> counter{0};
    for (auto _ : state) {
        counter.fetch_add(1, std::memory_order_acq_rel);
        benchmark::DoNotOptimize(counter);
    }
}
BENCHMARK(BM_AtomicIncrement_AcqRel);

static void BM_AtomicIncrement_SeqCst(benchmark::State& state) {
    std::atomic<int> counter{0};
    for (auto _ : state) {
        counter.fetch_add(1, std::memory_order_seq_cst);
        benchmark::DoNotOptimize(counter);
    }
}
BENCHMARK(BM_AtomicIncrement_SeqCst);

// 多线程竞争测试
static void BM_AtomicWithContention(benchmark::State& state) {
    const int kNumThreads = 4;
    std::atomic<int> counter{0};
    std::vector<std::thread> threads;
    for (auto _ : state) {
        counter.store(0, std::memory_order_relaxed);
        threads.clear();
        for (int i = 0; i < kNumThreads; ++i) {
            threads.emplace_back([&counter, &state]() {
                for (int j = 0; j < state.range(0); ++j) {
                    counter.fetch_add(1, std::memory_order_relaxed);
                }
            });
        }
        for (auto& t : threads) {
            t.join();
        }
        benchmark::DoNotOptimize(counter);
    }
}
BENCHMARK(BM_AtomicWithContention)->Arg(1000)->Arg(10000);
BENCHMARK_MAIN();

8. 常见误区与陷阱

8.1 错误的内存序组合

// 错误示例：不匹配的内存序
std::atomic<bool> flag{false};
int data = 0;

void thread1() {
    data = 42;
    flag.store(true, std::memory_order_release); // 释放
}

void thread2() {
    // 错误：应该使用 acquire 来匹配 release
    if (flag.load(std::memory_order_relaxed)) { // 应该是 acquire!
        std::cout << data << std::endl; // 可能看不到 42
    }
}

8.2 过度使用顺序一致性

// 不必要的性能开销
class Config {
    std::atomic<bool> updated{false};
    std::string value;

public:
    void update(const std::string& new_value) {
        value = new_value;
        // 过度使用：其实只需要 release
        updated.store(true, std::memory_order_seq_cst);
    }

    std::string get() {
        // 过度使用：其实只需要 acquire
        while (!updated.load(std::memory_order_seq_cst)) {}
        return value;
    }
};

8.3 忘记同步非原子数据

// 危险：非原子数据没有正确同步
struct Data {
    int a, b, c;
};
std::atomic<Data*> ptr{nullptr};

void writer() {
    Data* p = new Data{1, 2, 3};
    // 问题：Data 的成员初始化可能没有 happens-before 关系
    ptr.store(p, std::memory_order_release);
}

void reader() {
    Data* p = ptr.load(std::memory_order_acquire);
    if (p) {
        // 可能看到部分初始化的数据！
        std::cout << p->a << ", " << p->b << ", " << p->c << std::endl;
    }
}

9. 流程图解析

9.1 内存序的 happens-before 关系

happens-before 关系：

• synchronizes-with
• 线程 A: store flag=true release
• 线程 A: 非原子写 data=42
• 线程 B: 非原子读 data
• 线程 B: load flag=true acquire
• synchronizes-with 关系

9.2 内存屏障效果图

• 线程 B（消费者）
• 线程 A（生产者）
• 同步点 Acquire 屏障
• 原子加载 flag=true
• 普通读操作 #1
• 普通读操作 #2
• 普通写操作 #2
• 普通写操作 #1
• Release 屏障
• 原子存储 flag=true

10. 总结与最佳实践

10.1 关键要点总结

1. 理解内存重排：现代处理器和编译器会重排内存操作以优化性能
2. 选择合适的内存序：不是所有场景都需要最强的 seq_cst
3. 配对使用：release 必须与 acquire 或 acq_rel 配对
4. 同步非原子数据：原子操作只保护自身，非原子数据需要正确的内存序来同步
5. 性能权衡：更强的内存序带来更大的性能开销

10.2 最佳实践指南

1. 默认使用 memory_order_seq_cst：在不确定时，使用默认值最安全
2. 逐步优化：先写正确的代码，再根据需要优化内存序
3. 使用现有模式：遵循已知的正确模式（如双重检查锁定）
4. 充分测试：内存序错误可能导致偶发性 bug，需要充分测试
5. 文档化：在复杂的内存序使用处添加注释

10.3 推荐工具

1. ThreadSanitizer：检测数据竞争
2. 模型检查工具：如 CDSChecker、Nidhugg
3. 性能分析器：分析不同内存序的性能影响

// 最后的建议：保持简单
class SimpleSafeCounter {
    std::atomic<int> count{0}; // 默认使用 seq_cst

public:
    void increment() {
        count.fetch_add(1); // 默认内存序，安全简单
    }

    int get() const {
        return count.load(); // 默认内存序，安全简单
    }
};

// 除非有明确的性能需求，否则保持简单是最好的策略

结语

C++ 的内存序是一个强大但复杂的特性。通过理解 6 种内存序的语义和适用场景，开发者可以在保证正确性的前提下，编写出高性能的并发程序。记住：正确性永远比性能更重要。只有在确实需要优化性能，并且完全理解内存序语义时，才应该使用非默认的内存序。