当临界区极短、竞争又极其激烈时，std::mutex 的开销往往超过它保护的那几行代码本身：一次未争用的加解锁约 20 纳秒，而一次争用导致的线程挂起与唤醒可能耗费数微秒。无锁（lock-free）数据结构把「谁先拿到锁」的仲裁下沉到 CPU 的原子指令，用 CAS 循环取代阻塞等待。代价是复杂度陡增——ABA、内存回收、内存序，每一个都能让一个看似正确的实现随机崩溃。本文在 https://plumephp.com/cpp-atomic-memory-order/ 的内存序基础之上，聚焦数据结构实现与工程落地。
一、无锁编程的基础与代价
1.1 三种进度保证
并发算法的进度保证由弱到强分为三级，务必先分清：
| 级别 | 含义 | 典型实现 |
|---|---|---|
| 阻塞（blocking） | 持锁线程挂起会阻塞其他线程 | std::mutex |
| 无锁（lock-free） | 任意时刻至少有一个线程能推进 | CAS 循环 |
| 无等待（wait-free） | 每个线程都能在有限步内完成 | 部分计数器、SPSC 队列 |
「无锁」并不等于「更快」，它保证的是系统整体不会因为某个线程被调度器暂停而停滞，这对实时系统与高优先级线程有实质意义。
1.2 硬件原语
所有无锁结构最终都构建在少数几条原子指令上：
#include <atomic>
#include <cstdint>
std::atomic<int> counter{0};
// 1. 读-改-写：fetch_add 返回旧值
int old = counter.fetch_add(1, std::memory_order_relaxed);
// 2. 比较并交换：仅当当前值等于 expected 时写入 desired
int expected = 5;
bool ok = counter.compare_exchange_strong(
expected, 6, std::memory_order_acq_rel);
// 3. 弱版本：可能伪失败，适合放在循环里
int exp2 = 5;
while (!counter.compare_exchange_weak(exp2, 7,
std::memory_order_acq_rel, std::memory_order_relaxed)) {
// 失败时 exp2 已被更新为当前值，可直接重试
}
在 x86 上，compare_exchange 编译为 lock cmpxchg，天然带全屏障；在 ARM/POWER 等弱内存模型上则会展开为 ldaxr/stlxr 循环，必须显式提供内存序。compare_exchange_weak 允许「伪失败」（值其实相等却返回 false），但在重试循环中它往往生成更优代码，因为无需内层循环。
二、无锁栈的实现
2.1 push 与 pop 的 CAS 循环
无锁栈是最简单的无锁结构，本质是「用 CAS 更新头指针」：
template <typename T>
class LockFreeStack {
struct Node { T value; Node* next; };
std::atomic<Node*> head_{nullptr};
public:
void push(T value) {
Node* n = new Node{std::move(value), nullptr};
n->next = head_.load(std::memory_order_relaxed);
while (!head_.compare_exchange_weak( // 失败则用新 head_ 重试
n->next, n, std::memory_order_release,
std::memory_order_relaxed)) {}
}
bool pop(T& out) {
Node* old = head_.load(std::memory_order_acquire);
while (old && !head_.compare_exchange_weak(
old, old->next, std::memory_order_acquire,
std::memory_order_relaxed)) {}
if (!old) return false;
out = std::move(old->value);
// 危险：old 可能已被重新推入（ABA），直接 delete 会 use-after-free
// delete old;
return true;
}
};
注意 push 用 release、pop 用 acquire：前者保证节点内容在链接前对其他线程可见，后者保证读到节点后能看到其内容。这是最经典的 release/acquire 配对。
2.2 ABA 问题
上面的 pop 有两个致命缺陷。第一个是 ABA 问题：线程 1 读到 head_ = A，被调度器挂起；线程 2 弹出 A、再压入 B、又弹出 B、重新压入 A；线程 1 恢复后 CAS 成功，但此时 A->next 已被改写，栈结构被破坏。
解法是为指针附加上单调递增的版本号，使「值相同」不再等同于「状态相同」：
struct TaggedPtr {
std::uintptr_t ptr = 0;
std::uint64_t tag = 0; // 单调递增版本号
};
// 把 ptr 与 tag 打包进 128 位，用 16 字节 CAS（x86-64 支持 cmpxchg16b）
class AtomicTaggedPtr {
std::atomic<TaggedPtr> v_{TaggedPtr{}};
public:
bool cas(TaggedPtr& expected, TaggedPtr desired) {
return v_.compare_exchange_weak(expected, desired,
std::memory_order_acq_rel);
}
TaggedPtr load() const { return v_.load(std::memory_order_acquire); }
void store(TaggedPtr p) { v_.store(p, std::memory_order_release); }
};
注意 16 字节原子的 compare_exchange 在部分平台上并非无锁，编译器会退化为内部加锁，此时应当检测：
static_assert(std::atomic<TaggedPtr>::is_always_lock_free,
"16 字节 CAS 在该平台非无锁，需改用 hazard pointer 方案");
三、无锁队列
3.1 Michael-Scott 队列
MS 队列是教科书级的 MPMC（多生产者多消费者）无锁队列，用两个原子指针 head 与 tail，配合「哑结点（dummy node）」简化边界处理：
#include <atomic>
#include <utility>
template <typename T>
class MSQueue {
struct Node {
T value;
std::atomic<Node*> next{nullptr};
};
std::atomic<Node*> head_;
std::atomic<Node*> tail_;
public:
MSQueue() {
Node* dummy = new Node{};
head_.store(dummy, std::memory_order_relaxed);
tail_.store(dummy, std::memory_order_relaxed);
}
void enqueue(T value) {
Node* n = new Node{std::move(value)};
Node* tail = tail_.load(std::memory_order_relaxed);
while (true) {
Node* next = tail->next.load(std::memory_order_acquire);
if (next == nullptr) {
if (tail->next.compare_exchange_weak( // 链接到尾部
next, n, std::memory_order_release,
std::memory_order_relaxed)) break;
} else {
tail_.compare_exchange_weak( // tail 落后，帮忙推进
tail, next, std::memory_order_release,
std::memory_order_relaxed);
}
}
tail_.compare_exchange_weak(n, n, std::memory_order_release,
std::memory_order_relaxed);
}
bool dequeue(T& out) {
Node* head = head_.load(std::memory_order_relaxed);
while (true) {
Node* tail = tail_.load(std::memory_order_relaxed);
Node* next = head->next.load(std::memory_order_acquire);
if (next == nullptr) return false; // 空队列
if (head == tail) { // tail 落后，帮忙推进
tail_.compare_exchange_weak(
tail, next, std::memory_order_release,
std::memory_order_relaxed);
} else {
out = std::move(next->value);
if (head_.compare_exchange_weak( // 摘除哑结点
head, next, std::memory_order_release,
std::memory_order_relaxed)) return true;
}
}
}
};
MS 队列的核心思想是 helping：当发现 tail 指针落后时，任何线程都可以帮忙把它推进，从而避免某个被挂起的线程拖住整个队列。代价是平均需要约 2 次 CAS 才能完成一次入队。
3.2 SPSC 环形缓冲区
如果生产者与消费者各只有一个，可以用无锁环形缓冲区，它不需要 CAS，仅靠 load/store 与内存序即可实现无等待：
#include <atomic>
#include <array>
template <typename T, std::size_t N>
class SpscRing {
static_assert((N & (N - 1)) == 0, "容量必须是 2 的幂");
std::array<T, N> buf_;
std::atomic<std::size_t> head_{0}; // 消费者位置
std::atomic<std::size_t> tail_{0}; // 生产者位置
public:
bool push(const T& v) {
std::size_t t = tail_.load(std::memory_order_relaxed);
std::size_t next = (t + 1) & (N - 1);
if (next == head_.load(std::memory_order_acquire))
return false; // 满
buf_[t] = v;
tail_.store(next, std::memory_order_release); // 发布
return true;
}
bool pop(T& out) {
std::size_t h = head_.load(std::memory_order_relaxed);
if (h == tail_.load(std::memory_order_acquire))
return false; // 空
out = buf_[h];
head_.store((h + 1) & (N - 1), std::memory_order_release);
return true;
}
};
SPSC 环形队列在实践中用途极广：音频处理、日志采集、高频交易的行情分发，几乎都用它。它也是少数真正「无等待」的结构之一。
四、安全内存回收
4.1 真正的难点
上面所有 delete 都被刻意省略了。原因是：当线程 A 刚从队列摘下节点、尚未释放时，线程 B 可能仍持有指向该节点的指针正在读取。判断「还有没有别人在用这个节点」是无锁编程最难的部分。三种主流方案对比如下：
| 方案 | 思路 | 回收延迟 | 每操作开销 | 代表库 |
|---|---|---|---|---|
| Hazard Pointer | 线程声明正在读的指针 | 立即可回收 | 每次读要发布 + 扫描 | folly、libcds |
| RCU | 读者零开销，写者延迟回收 | 宽限期后 | 读者零，写者高 | Linux 内核、liburcu |
| Epoch-Based | 全局 epoch 计数 + 退役链表 | 跨过 epoch 后 | 很低 | crossbeam、folly |
4.2 Hazard Pointer 最小实现
Hazard pointer 的核心是：每个线程有一个公开的「危险指针」槽位，声明自己正在访问哪个节点；回收者遍历所有槽位，只释放无人声明的节点。
template <typename T>
class HazardPointer {
static constexpr int MAX_THREADS = 64;
static constexpr int MAX_HAZARDS = 2; // 每线程最多声明 2 个
std::atomic<T*> hazards_[MAX_THREADS][MAX_HAZARDS];
std::mutex retire_mtx_;
std::vector<T*> retired_;
public:
void protect(int tid, int slot, T* p) {
hazards_[tid][slot].store(p, std::memory_order_seq_cst);
}
void clear(int tid, int slot) {
hazards_[tid][slot].store(nullptr, std::memory_order_release);
}
// 回收时判断 p 是否被任何线程声明
bool is_hazardous(T* p) {
for (int t = 0; t < MAX_THREADS; ++t)
for (int s = 0; s < MAX_HAZARDS; ++s)
if (hazards_[t][s].load(std::memory_order_seq_cst) == p)
return true;
return false;
}
void retire(T* p) {
std::lock_guard<std::mutex> lk(retire_mtx_);
retired_.push_back(p);
for (auto it = retired_.begin(); it != retired_.end(); ) { // 尝试回收
if (!is_hazardous(*it)) { delete *it; it = retired_.erase(it); }
else ++it;
}
}
};
生产级实现（如 folly 的 HazptrDomain）会用批量回收、按域划分、以及「危险指针阈值」来降低扫描成本，但骨架思想与上面一致。
4.3 Epoch-Based Reclamation 简介
Epoch 方案把所有线程划分为「活跃（active）」与「静止（quiescent）」两种状态，并维护一个全局 epoch 计数器。节点退役时挂到当前 epoch 的链表上；当所有线程都跨过至少一个 epoch 后，旧 epoch 的链表即可整体释放。它比 hazard pointer 的每次读开销更低，但要求线程周期性进入静止态，否则会导致内存无限堆积。
五、内存序选型与压测验证
5.1 内存序选择清单
无锁结构的正确性一半取决于内存序。经验法则：
- 发布数据（写指针后让别人看到内容）：
release - 订阅数据（读指针后访问内容）：
acquire - 纯计数、无数据关联：
relaxed - CAS 需要同时同步两侧：
acq_rel - 实在不确定：先用
seq_cst保证正确，再用perf与压力测试逐步放宽
// 反例：用 relaxed 发布节点，其他线程可能读到未初始化的 value
node->next.store(old, std::memory_order_relaxed);
head_.store(node, std::memory_order_relaxed); // 错误：无发布语义
// 正确：store-release 建立 happens-before
head_.store(node, std::memory_order_release);
5.2 压测与验证方法
无锁 bug 的特征是「低概率、依赖调度、只在特定 CPU 上复现」，因此必须用专门的验证手段：
- 高并发压测：线程数远超核心数（如 8 核跑 64 线程）制造频繁抢占
- TSan（ThreadSanitizer）：
-fsanitize=thread能捕获大部分数据竞争，但对无锁代码会报告「benign race」，需用__tsan_acquire/__tsan_release注解 - 长时间运行：至少跑 10^8 次操作，配合
taskset绑定 CPU 制造争用 - 不变量检查：每 N 次操作后单线程遍历结构，校验节点数、无环、无重复
- CPU 架构差异：x86 的强内存模型会掩盖 bug，务必在 ARM 机器或 QEMU 上验证
// 压测骨架：多生产者多消费者 + 计数校验
void stress_test(MSQueue<int>& q, int producers, int consumers) {
std::atomic<long> produced{0}, consumed{0};
constexpr int PER_THREAD = 1'000'000;
std::vector<std::thread> ts;
for (int i = 0; i < producers; ++i)
ts.emplace_back([&] { for (int k = 0; k < PER_THREAD; ++k)
{ q.enqueue(k); produced.fetch_add(1); } });
for (int i = 0; i < consumers; ++i)
ts.emplace_back([&] { int v;
while (consumed.load() < (long)producers * PER_THREAD)
if (q.dequeue(v)) consumed.fetch_add(1); });
for (auto& t : ts) t.join();
assert(produced.load() == consumed.load());
}
六、工程建议与陷阱
无锁数据结构是把双刃剑，落地前请权衡以下几点：
- 先测量再决定：多数场景下，
std::mutex配合短临界区已经足够快；无锁只在 profile 证明锁是瓶颈时才值得引入 - 优先用成熟库：
boost::lockfree、folly::ProducerConsumerQueue、moodycamel::ConcurrentQueue都经过大规模生产验证，自研风险极高 - ABA 与回收是两大雷区：没有版本号或安全回收方案的无锁结构，几乎必然在某次线上高并发时崩溃
- 弱内存模型平台必须实测：x86 上正确的代码在 ARM 上可能因缺少屏障而失效
- 不要在无锁结构里做重活：CAS 循环内只做指针操作，任何可能阻塞或耗时的调用都会放大重试成本
与 https://plumephp.com/cpp-concurrency-patterns/ 中的线程池结合时，SPSC 环形队列常被用作任务提交队列，既避免了锁，又天然契合「单生产者提交、单消费者执行」的模型。
相关阅读
- https://plumephp.com/cpp-atomic-memory-order/ — 内存序模型、happens-before 与 ABA 问题的基础
- https://plumephp.com/cpp-concurrency-patterns/ — 线程池、future 与异步任务调度模式
- https://plumephp.com/cpp-threading-basics/ — 线程生命周期与 mutex/条件变量的正确用法
延伸阅读
- https://plumephp.com/posts/os/ — 内核中的 RCU、自旋锁与调度器如何影响无锁代码
- https://plumephp.com/posts/hpc/ — 大规模并行计算中的无锁通信与原子操作实践
文末完整示例
// 完整可运行示例：无等待 SPSC 环形队列 + 并发校验
// 编译：g++ -std=c++20 -O2 -pthread -o lockfree_demo lockfree_demo.cpp
#include <atomic>
#include <array>
#include <thread>
#include <iostream>
template <typename T, std::size_t N>
class SpscRing {
static_assert((N & (N - 1)) == 0, "容量必须是 2 的幂");
std::array<T, N> buf_{};
std::atomic<std::size_t> head_{0};
std::atomic<std::size_t> tail_{0};
public:
bool push(const T& v) {
std::size_t t = tail_.load(std::memory_order_relaxed);
std::size_t next = (t + 1) & (N - 1);
if (next == head_.load(std::memory_order_acquire)) return false;
buf_[t] = v;
tail_.store(next, std::memory_order_release);
return true;
}
bool pop(T& out) {
std::size_t h = head_.load(std::memory_order_relaxed);
if (h == tail_.load(std::memory_order_acquire)) return false;
out = buf_[h];
head_.store((h + 1) & (N - 1), std::memory_order_release);
return true;
}
};
int main() {
SpscRing<int, 1024> ring;
constexpr int TOTAL = 1'000'000;
long sum = 0;
std::thread producer([&] {
for (int i = 1; i <= TOTAL; )
if (ring.push(i)) ++i;
});
std::thread consumer([&] {
int v, got = 0;
while (got < TOTAL)
if (ring.pop(v)) { sum += v; ++got; }
});
producer.join();
consumer.join();
long expect = (long)TOTAL * (TOTAL + 1) / 2;
std::cout << "SPSC 校验：sum=" << sum << " 期望=" << expect
<< (sum == expect ? " OK" : " FAIL") << std::endl;
return 0;
}
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。