C++ 高性能并发编程：基于 RAII 与 CAS 的轻量级原子读写锁设计

2. 为什么要做这样的分层设计？

你可能会问，为什么不直接在锁的类里面写死这些逻辑？这种基于模板的泛型分层设计有三大绝佳优势：

1. 极致的解耦与复用（Duck Typing）：这里的 Guard 并不关心底层的锁是怎么实现的。无论你是基于 std::mutex 自己封装的锁，还是基于 pthread_rwlock_t 的系统锁，亦或是我们接下来要写的原子锁，只要这个类包含了 ReadLock() 等方法，就可以直接套用这套 Guard。
2. 零成本抽象（Zero-cost Abstraction）：因为使用了模板，编译器会在编译期进行类型推导和内联展开，没有任何虚函数调用的开销，运行时性能损耗为零。
3. 职责单一原则：Guard 只负责生命周期管理（防死锁），而底层的 Lock 类只负责并发控制。代码结构清晰，极易维护。

二、核心引擎：基于 CAS 的原子读写锁（深度解析）

有了安全的 RAII 外壳，接下来我们来实现硬核的底层锁：AtomicRWLock。

1. 为什么要抛弃传统锁？（Why Lock-free?）

传统的 std::mutex 或 std::shared_mutex 底层通常依赖操作系统的同步原语（如 Linux 的 futex）。当锁竞争失败时，线程会被挂起（Sleep），这涉及用户态到内核态的切换以及线程上下文切换。这个过程的开销通常在微秒级（us）。

在极高并发或超低延迟（纳秒级，ns）的场景下，这种开销是不可接受的。因此，我们需要一种完全在用户态解决竞争的方案，这就是基于 std::atomic 的无锁化（Lock-free）设计。

2. 状态机的精妙设计：为什么只用一个变量？

在无锁编程中，最核心的限制是：CPU 的 CAS（Compare-And-Swap）指令通常只能保证对单个内存地址的原子操作。

因此，我们不能用一个变量表示'是否有写锁'，再用另一个变量表示'有多少个读锁'，因为你无法同时原子地修改这两个变量。

我们的解决方案是：将所有状态压缩到一个 std::atomic<int32_t> lock_num_ 中：

• 0：空闲状态（RW_LOCK_FREE）。
• -1：被写线程独占（WRITE_EXCLUSIVE）。
• > 0：正在被多个读线程共享，其数值代表当前持有读锁的线程数量。

通过这种设计，无论是加读锁（+1）还是加写锁（变成 -1），都可以通过一次 CAS 操作完成，保证了绝对的原子性。

3. 解决'写饥饿'：写优先策略

读写锁最容易遇到的坑是写饥饿：如果读线程源源不断地到来，lock_num_ 一直大于 0，写线程可能永远也拿不到锁。

为此，我们引入了辅助变量 write_lock_wait_num_。当写线程准备获取锁时，先将此变量 +1。读线程在尝试加锁前，必须先检查这个变量。如果发现有写线程在等，读线程就会主动放弃竞争，把机会让给写线程。

4. CAS 循环与自旋退让（Yield）的艺术

我们来看看加写锁的核心逻辑：

// 尝试将状态从 0 修改为 -1
while (!lock_num_.compare_exchange_weak(rw_lock_free, WRITE_EXCLUSIVE, ...)) {
    rw_lock_free = RW_LOCK_FREE; // CAS 失败会修改 expected 值，需重置
    if (++retry_times > MAX_RETRY_TIMES) {
        retry_times = 0;
        std::this_thread::yield(); // 避免死自旋占满 CPU
    }
}

为什么用 compare_exchange_weak 而不是 strong？

在某些架构（如 ARM）上，weak 版本可能会出现'伪失败'（即使值相等也返回 false），但它的指令开销更小。因为我们本来就在一个 while 循环中，伪失败大不了再循环一次，所以用 weak 性能更高。

为什么要 yield？

如果单纯使用 while 死循环（纯自旋锁），当持有锁的线程被操作系统调度走时，等待锁的线程会 100% 占满 CPU 核心进行无意义的空转。引入 std::this_thread::yield() 后，如果重试多次仍未拿到锁，线程会主动告诉 OS：'我先放弃当前的时间片，你去执行别的线程吧'。这是一种极佳的性能折中。

5. 细节拉满：内存序（Memory Order）的底层逻辑

在并发编程中，比死锁更可怕的是指令重排。编译器和 CPU 为了优化性能，可能会打乱代码的执行顺序。如果'读取共享数据'的指令被重排到了'获取锁'的指令之前，锁就形同虚设了。

因此，我们不能使用默认的 std::memory_order_seq_cst（全局顺序一致性，开销最大，会清空 CPU 流水线），而是精准地使用了 acquire-release 语义：

• 加锁操作（Acquire）：load 和 CAS 成功时使用 memory_order_acquire。它像一道单向栅栏，确保加锁之后的读写操作，绝不会被重排到加锁之前。
• 解锁操作（Release）：fetch_add/sub 解锁时使用 memory_order_release。它确保解锁之前的所有数据修改，都已经同步到主存，对其他线程可见。

这种精细的内存序控制，能将 CPU 缓存一致性协议（如 MESI）的同步开销降到最低，榨干硬件的最后一丝性能。

6. 完整源码实现

为了让你更好地理解上述所有细节是如何协同工作的，这里给出 AtomicRWLock 的完整实现代码：

// atomic_rw_lock.h
#include <atomic>
#include <thread>
#include "rw_lock_guard.h"

namespace cmw {
namespace base {
class AtomicRWLock {
    // 允许 Guard 类访问私有的加解锁接口，对外隐藏底层细节
    friend class ReadLockGuard<AtomicRWLock>;
    friend class WriteLockGuard<AtomicRWLock>;

public:
    static constexpr int32_t RW_LOCK_FREE = 0;
    static constexpr int32_t WRITE_EXCLUSIVE = -1;
    static constexpr uint32_t MAX_RETRY_TIMES = 5;

    AtomicRWLock() = default;
    explicit AtomicRWLock(bool write_first) : write_first_(write_first) {}

private:
    std::atomic<uint32_t> write_lock_wait_num_ = {0};
    std::atomic<int32_t> lock_num_ = {0};
    bool write_first_ = true;

    void ReadLock() {
        uint32_t retry_times = 0;
        int32_t lock_num = lock_num_.load(std::memory_order_acquire);
        if (write_first_) {
            do {
                // 写优先：如果有写线程在等待，读线程必须阻塞
                while (lock_num < RW_LOCK_FREE || write_lock_wait_num_.load(std::memory_order_acquire) > 0) {
                    if (++retry_times > MAX_RETRY_TIMES) {
                        retry_times = 0;
                        std::this_thread::yield(); // 避免死自旋占满 CPU
                    }
                    lock_num = lock_num_.load(std::memory_order_acquire);
                }
            } while (!lock_num_.compare_exchange_weak(lock_num, lock_num + 1, std::memory_order_acq_rel, std::memory_order_relaxed));
        } else {
            // 读优先：只要没有写线程正在占用（lock_num < 0），读线程就可以获取锁
            do {
                while (lock_num < RW_LOCK_FREE) {
                    if (++retry_times > MAX_RETRY_TIMES) {
                        retry_times = 0;
                        std::this_thread::yield();
                    }
                    lock_num = lock_num_.load(std::memory_order_acquire);
                }
            } while (!lock_num_.compare_exchange_weak(lock_num, lock_num + 1, std::memory_order_acq_rel, std::memory_order_relaxed));
        }
    }

    void WriteLock() {
        uint32_t retry_times = 0;
        int32_t rw_lock_free = RW_LOCK_FREE;
        // 登记：有一个写线程正在等待
        write_lock_wait_num_.fetch_add(1, std::memory_order_release);
        // 尝试将状态从 0 修改为 -1
        while (!lock_num_.compare_exchange_weak(rw_lock_free, WRITE_EXCLUSIVE, std::memory_order_acq_rel, std::memory_order_relaxed)) {
            rw_lock_free = RW_LOCK_FREE; // CAS 失败会修改 expected 值，需重置
            if (++retry_times > MAX_RETRY_TIMES) {
                retry_times = 0;
                std::this_thread::yield();
            }
        }
        // 获取成功，注销等待状态
        write_lock_wait_num_.fetch_sub(1, std::memory_order_release);
    }

    void ReadUnlock() {
        lock_num_.fetch_sub(1, std::memory_order_release);
    }

    void WriteUnlock() {
        lock_num_.fetch_add(1, std::memory_order_release);
    }
};
} // namespace base
} // namespace cmw

三、优雅的使用方式

得益于分层设计，在业务代码中使用这套高并发组件变得异常简单且安全：

cmw::base::AtomicRWLock my_lock(true); // 开启写优先
int shared_data = 0;

void ReaderThread() {
    // 作用域开始，自动加读锁
    cmw::base::ReadLockGuard<cmw::base::AtomicRWLock> guard(my_lock);
    std::cout << "Read data: " << shared_data << std::endl;
    // 作用域结束，自动解读锁
}

void WriterThread() {
    // 作用域开始，自动加写锁
    cmw::base::WriteLockGuard<cmw::base::AtomicRWLock> guard(my_lock);
    shared_data++; // 作用域结束，自动解写锁
}

总结

通过这篇文章，我们实现了一个工业级的高性能读写锁。回顾整个过程，我们做对了三件事：

1. 用 RAII 兜底：彻底消灭了忘记解锁导致的死锁隐患。
2. 面向接口的泛型编程：Guard 与 Lock 解耦，任何实现了对应接口的锁都能无缝接入。
3. 无锁化与写优先：利用 CAS 和精细的内存序榨干 CPU 性能，同时通过 yield 和等待计数器兼顾了系统的公平性。

希望这种分层设计与底层优化的思路，能对你的日常开发有所启发！