C++20 原子共享指针：当并发遇上“分披萨”的艺术
各位听众，大家好！
欢迎来到今天的研讨会。我是你们的讲师，一个在代码泥潭里摸爬滚打多年，头发比发际线后退得还慢的资深 C++ 程序员。今天，我们要聊一个极其性感、极其硬核，甚至有点“反直觉”的话题。
在 C++ 的世界里，std::shared_ptr 是个老朋友。它就像披萨店里的那个大圆盘，大家都可以往上面放片肉，甚至可以分着吃。但是，如果在多线程的高压环境下——比如有成千上万个厨师（线程）同时在这个圆盘上操作——这个老朋友就会变得极其暴躁，甚至会把厨房炸了。
这就是我们要解决的问题：如何在并发场景下，安全、高效地共享对象所有权？
在 C++20 之前，我们只能用互斥锁（std::mutex）或者读写锁（std::shared_mutex）。但今天，C++20 给我们送来了“核武器”：std::atomic<std::shared_ptr>。
别急着翻书，这玩意儿不是魔法，它是基于内存模型和硬件指令的精密艺术。今天，我们就来聊聊如何利用它构建一个高性能的无锁并发对象共享协议。
第一部分：为什么 std::shared_ptr 在并发中是个“疯子”？
在深入 C++20 之前，我们必须先理解为什么以前的 shared_ptr 在多线程下是个定时炸弹。
想象一下，线程 A 想要读取一个 shared_ptr 指向的数据，线程 B 想要释放它。
在传统的 C++ 中，shared_ptr 的内部逻辑是这样的：
- 拷贝构造：增加引用计数（
use_count）。 - 析构：减少引用计数。如果计数降为 0，销毁对象。
现在，请闭上眼睛想象一下这个场景：
- 线程 A 执行了
auto p = global_ptr;。它拿到了指针，引用计数变成了 2。 - 线程 B 执行了
global_ptr.reset();。它执行了--use_count，计数变成了 1。它以为对象还活着，于是把指针置空了。 - 线程 A 刚想访问
p->doSomething()，结果p是个空指针。程序崩溃。
这仅仅是“读取”和“释放”的简单冲突。如果你加上“修改数据”和“销毁对象”，那简直就是地狱。传统的做法是什么？加锁。std::shared_mutex。这就像大家在分披萨的时候，必须排成一队，一个一个来拿。
问题来了：
- 性能瓶颈：互斥锁会触发上下文切换，CPU 需要从用户态切换到内核态，这太慢了！
- 饿死现象：如果有一个线程一直持有写锁，其他想读的线程就得干等着。
- 缓存行失效：锁操作会导致 CPU 缓存频繁失效，内存带宽被打爆。
我们渴望一种不需要排队、不需要等待、不需要锁的方案。C++20 的 std::atomic<std::shared_ptr> 就是这样一位“极速侠”。
第二部分：C++20 的魔法——std::atomic<std::shared_ptr>
那么，std::atomic<std::shared_ptr> 到底做了什么神奇的事情？
它并没有把 shared_ptr 的内部指针变成原子的（那是做不到的，因为指针本身不是 trivially copyable 的，而且引用计数也不是原子的），而是把 shared_ptr 的控制块 变成了原子的。
当你创建一个 std::atomic<std::shared_ptr<T>> 时，它实际上是一个包装器。当你调用 store 或 load 时，它会对底层的控制块（__shared_ptr_control_block）执行原子的 compare_exchange 操作。
这意味着什么？
这意味着，无论多少个线程同时对这个原子指针进行读写，C++ 标准库保证你看到的 shared_ptr 状态要么是“旧状态”，要么是“新状态”，绝不可能是一个残缺的、中间态。
核心特性：原子操作接口
C++20 为 std::atomic<std::shared_ptr> 提供了一整套标准的原子操作接口。这不仅仅是 load 和 store，还包括了：
compare_exchange_weak/compare_exchange_strong：用于无锁算法中的 CAS（Compare-And-Swap）操作。exchange：原子地交换值。fetch_add/fetch_sub：虽然shared_ptr没有直接的“增加引用计数”操作（那是隐式的），但这个接口的存在暗示了底层的原子性控制。
第三部分：构建“无锁共享协议”
光说不练假把式。我们要构建一个协议，这个协议要解决三个核心问题：
- 所有权转移：谁能拿走这个对象？
- 并发读取：谁能看这个对象？
- 生命周期管理：对象啥时候死？
让我们来设计一个简单的 ConcurrentObjectPool（并发对象池）来演示这个协议。
协议规则
- 生产者：创建对象，存入原子指针。
- 消费者：调用
acquire()，原子地获取所有权。 - 观察者：调用
observe()，原子地读取指针但不获取所有权。 - 释放者：调用
release()，原子地减少引用计数。
代码实现：基础版
让我们看一段代码。注意，这里没有 std::mutex，没有 std::lock_guard，只有纯粹的原子操作。
#include <atomic>
#include <memory>
#include <iostream>
#include <thread>
#include <vector>
#include <chrono>
// 假设这是一个昂贵的资源，比如数据库连接
class ExpensiveResource {
public:
ExpensiveResource(int id) : id_(id) {
std::cout << "Resource " << id_ << " created." << std::endl;
}
~ExpensiveResource() {
std::cout << "Resource " << id_ << " destroyed." << std::endl;
}
void doWork() {
std::cout << "Resource " << id_ << " is working..." << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
private:
int id_;
};
// 我们要构建的协议类
class AtomicResourceManager {
public:
AtomicResourceManager() : atomic_ptr_(nullptr) {}
// 1. 生产：创建资源并放入原子指针
void produce(int id) {
auto new_resource = std::make_shared<ExpensiveResource>(id);
// store 是原子的，底层会保证控制块的原子性
atomic_ptr_.store(new_resource, std::memory_order_relaxed);
}
// 2. 消费：获取所有权
// 我们使用 compare_exchange_strong 来尝试交换
std::shared_ptr<ExpensiveResource> acquire() {
std::shared_ptr<ExpensiveResource> old_ptr;
// 循环直到成功获取所有权
while (true) {
// 1. 先加载当前状态
old_ptr = atomic_ptr_.load(std::memory_order_acquire);
// 2. 尝试 CAS：如果当前原子指针还是 old_ptr，则替换为 nullptr
// 我们把所有权“偷”走了，所以放空
if (atomic_ptr_.compare_exchange_strong(
old_ptr,
nullptr,
std::memory_order_acq_rel, // 释放语义
std::memory_order_acquire // 获取语义
)) {
// CAS 成功！我们拿到了所有权
return old_ptr;
}
// CAS 失败，说明有别的线程捷足先登了，继续循环
}
}
// 3. 释放：把对象还回去（或者减少引用计数）
void release(std::shared_ptr<ExpensiveResource> ptr) {
if (ptr) {
// 把指针放回去，shared_ptr 的拷贝构造会自动增加引用计数
atomic_ptr_.store(ptr, std::memory_order_release);
}
}
// 4. 观察：只读，不拿所有权
std::shared_ptr<ExpensiveResource> observe() const {
// 只需要 acquire 语义，因为我们只是读取，不修改
return atomic_ptr_.load(std::memory_order_acquire);
}
private:
std::atomic<std::shared_ptr<ExpensiveResource>> atomic_ptr_;
};代码解析：
看 acquire() 函数，这简直是无锁算法的教科书。
- 我们声明一个
old_ptr。 load获取当前值。compare_exchange_strong尝试将原子指针从old_ptr改为nullptr。- 关键点：如果 CAS 成功，意味着我们成功抢到了所有权（原来的指针变成了
nullptr），返回old_ptr。 - 如果 CAS 失败，说明有其他线程也调用了
acquire()，或者有生产者放回了新对象。这时我们重新load，再次尝试。
这里用到了内存顺序。std::memory_order_acq_rel 用于 CAS 操作，确保了读取到的对象状态是完整的，并且保证了释放操作对后续获取操作的可见性。
第四部分：进阶协议——支持多观察者的“读写锁替代品”
上面的例子有点简单，因为生产者一旦生产，消费者一抢就没了，没人能观察。在现实世界中，我们经常需要：“大家都想看，但没人能销毁”。
这就需要更复杂的协议。我们需要区分“独占访问”（所有权）和“共享访问”（只读）。
让我们实现一个高级的 ConcurrentDocument，就像一个多人在线文档编辑器。
核心思想：CAS + 原子计数
我们将 std::atomic<std::shared_ptr> 的 use_count 视为“观察者数量”。
class ConcurrentDocument {
public:
ConcurrentDocument() {
// 初始化一个空的控制块
// 实际上 shared_ptr 的默认构造会创建一个空的 shared_ptr
atomic_ptr_.store(nullptr, std::memory_order_relaxed);
}
// 写操作：获取独占所有权
// 返回 true 表示成功，false 表示文档已被锁定
bool try_lock_for_write() {
std::shared_ptr<Content> old_content;
while (true) {
old_content = atomic_ptr_.load(std::memory_order_acquire);
// 如果指针为空，说明没人拥有，也没人观察，我们可以创建新内容
if (!old_content) {
auto new_content = std::make_shared<Content>();
if (atomic_ptr_.compare_exchange_strong(old_content, new_content,
std::memory_order_acq_rel, std::memory_order_acquire)) {
return true;
}
continue;
}
// 如果指针不为空，检查观察者数量
// use_count = 1 表示只有拥有者自己，我们可以独占
if (old_content.use_count() == 1) {
// 尝试将 use_count 降为 0，同时把所有权转移给当前线程
// 这需要手动操作控制块，比较复杂。
// 为了简化，我们这里演示一种更“Hack”但常用的技巧：
// 我们不修改 use_count，而是通过 exchange 交换指针。
// 但这样会破坏 shared_ptr 的语义。
// 所以，真正的无锁读写锁实现通常需要自定义控制块。
// 这里为了代码可读性，我们演示一种简化版逻辑：
// 真实场景下，这里会通过 CAS 修改控制块的引用计数
// 或者使用 atomic_ref 修改引用计数
}
return false; // 被锁定
}
}
// 读操作：共享访问
// 只要文档存在，任何人都可以读
std::shared_ptr<Content> get_content() {
// 直接 load，shared_ptr 的拷贝会增加 use_count
return atomic_ptr_.load(std::memory_order_acquire);
}
// 释放写操作
void unlock() {
// shared_ptr 的析构会自动减少 use_count
// 但由于它是 shared_ptr，析构并不会销毁对象，除非 use_count 为 0
// 这里我们不需要做任何事，只要不持有指针即可。
// 或者，如果我们想释放所有权，可以 store nullptr
atomic_ptr_.store(nullptr, std::memory_order_release);
}
};
class Content {
// ...
};等等，上面的代码有点偷懒。 让我们更深入一点。真正的 std::atomic<std::shared_ptr> 的威力在于，我们可以利用它来管理控制块本身。
真正的无锁协议：基于 std::atomic_ref 的引用计数
要实现一个完美的无锁读写锁，我们不能只依赖 std::atomic<std::shared_ptr> 的 load/store，因为 shared_ptr 的拷贝构造和析构函数是隐式的，我们很难在不加锁的情况下精确控制引用计数的增减。
但是，C++20 引入了 std::atomic_ref，这简直是神器！
我们可以把 shared_ptr 的控制块中的 use_count 字段提取出来，用 std::atomic_ref 包装，进行原子的增减。
这里是一个更硬核的实现思路（伪代码演示）：
// 假设 shared_ptr 的控制块结构大致如下
struct SharedControlBlock {
std::atomic<int> use_count;
std::atomic<int> weak_count;
T* data_ptr;
};
template <typename T>
class LockFreeSharedPtr {
public:
LockFreeSharedPtr() : control_block_(nullptr) {}
// 构造：原子地创建控制块
LockFreeSharedPtr(T* ptr) : control_block_(nullptr) {
auto* block = new SharedControlBlock();
block->data_ptr = ptr;
block->use_count.store(1, std::memory_order_relaxed);
// 原子地存储控制块指针
control_block_.store(block, std::memory_order_release);
}
// 拷贝构造：原子地增加引用计数
LockFreeSharedPtr(const LockFreeSharedPtr& other) noexcept {
// 1. 原子地加载对方的控制块
auto* block = other.control_block_.load(std::memory_order_acquire);
if (block) {
// 2. 原子地增加计数
// 使用 fetch_add，因为它比 compare_exchange 更适合单纯的计数
block->use_count.fetch_add(1, std::memory_order_acq_rel);
// 3. 原子地存储到当前对象
control_block_.store(block, std::memory_order_release);
}
}
// 赋值：原子的“原子”操作
LockFreeSharedPtr& operator=(const LockFreeSharedPtr& other) noexcept {
if (this != &other) {
// 先增加对方的计数
auto* other_block = other.control_block_.load(std::memory_order_acquire);
if (other_block) {
other_block->use_count.fetch_add(1, std::memory_order_acq_rel);
}
// 再减少当前的计数
auto* old_block = control_block_.load(std::memory_order_acquire);
if (old_block) {
// 如果计数降为 0，销毁对象
if (old_block->use_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
delete old_block;
}
}
// 更新当前指针
control_block_.store(other_block, std::memory_order_release);
}
return *this;
}
// 获取裸指针
T* get() const {
auto* block = control_block_.load(std::memory_order_acquire);
return block ? block->data_ptr : nullptr;
}
private:
std::atomic<SharedControlBlock*> control_block_;
};这段代码说明了什么？
它说明，我们可以完全绕过 std::shared_ptr 的封装，直接利用 std::atomic 对内存进行操作。这就是 std::atomic<std::shared_ptr> 能够存在的原因——它把 shared_ptr 视为一个原子变量，而它的值（控制块指针）是原子的。
第五部分：性能分析——为什么这比锁快？
好了，我们现在有了代码，也知道了原理。但为什么这比 std::shared_mutex 快？
1. 消除了上下文切换
互斥锁在竞争激烈时，操作系统需要挂起线程，切换 CPU 核心去执行其他任务。这涉及用户态到内核态的切换，成本大概是几百个 CPU 周期。
原子操作是在 CPU 的缓存行上进行的，纯粹是 CPU 指令层面的操作。虽然原子操作在竞争时也会忙等待，但这只是 CPU 在空转，没有系统调用的开销。
2. 缓存友好性
互斥锁通常有一个全局的互斥量对象，所有线程都在抢夺这个对象的锁标志位。这会导致大量的缓存行伪共享。
而 std::atomic<std::shared_ptr> 每个线程持有的数据结构（控制块指针）通常在堆上，或者分散在内存的不同位置，减少了缓存冲突。
3. 细粒度控制
互斥锁是粗粒度的。一旦锁住，谁都不能动。
原子操作是细粒度的。线程 A 读，线程 B 写，只要它们操作的是不同的内存地址（比如不同的控制块），它们可以同时进行。
4. 避免了锁的“优先级反转”
在实时系统中，互斥锁会导致优先级反转：低优先级的线程持有锁，高优先级的线程在等待。原子操作通过忙等待（或者硬件提供的 compare_exchange）通常不会阻塞线程，从而避免了这个问题（当然，忙等待会占用 CPU）。
第六部分：陷阱与最佳实践——不要在真空中使用原子
虽然 std::atomic<std::shared_ptr> 很强大，但作为专家，我必须警告你：不要滥用！
陷阱一：ABA 问题
在无锁编程中，compare_exchange 经常面临 ABA 问题：值从 A 变成了 B，又变回了 A。
对于 std::atomic<std::shared_ptr>，这个问题相对少见，因为 shared_ptr 的控制块指针通常不会频繁地被释放再分配（除非是内存池）。但在复杂的自定义协议中，你需要小心。
陷阱二：内存顺序的选择
不要默认使用 std::memory_order_relaxed！
在 acquire-release 模式中，你必须使用 memory_order_acquire 来读取，memory_order_release 来写入。如果你用错了顺序，可能会导致数据竞争，或者更糟糕的是，编译器或 CPU 优化掉你的内存屏障，导致数据不一致。
陷阱三：std::shared_ptr 的开销
std::shared_ptr 需要分配堆内存来存储控制块。如果你在一个高频循环中创建和销毁 std::atomic<std::shared_ptr>，内存分配器的开销可能会抵消原子操作带来的性能提升。
最佳实践：尽量重用控制块，或者使用 std::unique_ptr（非原子）配合 std::atomic 来管理所有权，只在需要共享时才复制控制块指针。
陷阱四：控制块大小
std::atomic<std::shared_ptr> 的值是控制块的指针。这意味着它占用的是指针大小（通常是 8 字节），而不是整个控制块的大小。这是一个巨大的优势！它把“重量级”的对象控制逻辑隐藏在了指针之后。
第七部分：实战案例——构建一个高性能的消息队列
让我们看看这个技术在实际场景中如何应用。假设我们有一个高性能的消息队列，多个生产者往里扔消息，多个消费者往外拿消息。
传统的做法是用一个队列加锁。
现在，我们用 std::atomic<std::shared_ptr> 来实现一个无锁队列。
核心思路：
每个消息是一个对象。队列中存储的是“链表节点”的原子指针。
生产者 CAS 插入节点。消费者 CAS 移除节点。
template <typename T>
class LockFreeQueue {
public:
struct Node {
T data;
Node* next;
};
LockFreeQueue() {
head_.store(nullptr, std::memory_order_relaxed);
tail_.store(nullptr, std::memory_order_relaxed);
}
void push(T value) {
auto* node = new Node{std::move(value), nullptr};
// 这是一个简化的无锁入队逻辑，实际生产环境需要更复杂的技巧
// 比如 Backoff, 或者使用 Michael-Scott 算法
Node* old_tail = tail_.load(std::memory_order_relaxed);
while (true) {
if (old_tail->next == nullptr) {
// 尝试将当前节点的 next 设置为 nullptr
if (old_tail->next.compare_exchange_strong(nullptr, node,
std::memory_order_relaxed, std::memory_order_relaxed)) {
// 成功插入，更新 tail
tail_.store(node, std::memory_order_release);
return;
}
} else {
// tail 落后了，帮它更新到链表末尾
tail_.compare_exchange_strong(old_tail, old_tail->next,
std::memory_order_relaxed, std::memory_order_relaxed);
}
// 重新加载 tail，因为 CAS 可能修改了它
old_tail = tail_.load(std::memory_order_relaxed);
}
}
bool pop(T& out_value) {
// 逻辑类似 push，CAS 移除 head 节点
auto* old_head = head_.load(std::memory_order_relaxed);
while (old_head) {
auto* next = old_head->next.load(std::memory_order_relaxed);
// 尝试更新 head
if (head_.compare_exchange_strong(old_head, next,
std::memory_order_relaxed, std::memory_order_relaxed)) {
out_value = std::move(old_head->data);
delete old_head; // 销毁节点
return true;
}
}
return false;
}
private:
std::atomic<Node*> head_;
std::atomic<Node*> tail_;
};在这个例子中，我们利用 std::atomic<Node*> 实现了链表节点的并发操作。虽然这是一个简化的实现（真正的 Michael-Scott 算法更复杂），但它完美展示了 std::atomic 如何管理共享状态。
结语：拥抱无锁，但保持清醒
各位朋友，C++20 的 std::atomic<std::shared_ptr> 是一把利剑。它允许我们在不牺牲性能的前提下，安全地在多线程间传递所有权和引用。
它不是万能的。如果你需要复杂的逻辑判断，或者需要遍历共享数据结构，传统的互斥锁依然是你的好朋友。但当你面对高吞吐、低延迟、数据量大的场景时，不要犹豫，试试原子共享指针。
记住，无锁编程不是关于“没有锁”，而是关于“没有阻塞”。它要求我们更深入地理解内存模型，更精细地控制硬件指令。
代码写得漂亮，就像写诗；代码写得无锁，就像在刀尖上跳舞。
好了，今天的讲座就到这里。现在，去编译你的代码，把那些互斥锁扔进垃圾桶吧！（当然，如果编译不过，记得先加上 #include <atomic>）。
谢谢大家！