C++无锁数据结构实现示例详解-北京尧图网络

前言无锁lock-free是并发编程里被误用最多的词之一。很多资料把它简化成不使用std::mutex于是把一段用std::atomic保护计数器的代码也称作无锁队列另一些资料则暗示无锁一定比加锁快。这两点都不准确。无锁描述的是进度保证progress guarantee不是某一种具体写法而无锁结构在低竞争时可能更快在高竞争时完全可能比一把短临界区的互斥量更慢。本文先讲清无锁、无等待、无阻塞这几个概念的区别再讲 CAS 与内存序这一对基础设施最后给出两个可以逐行推演的实现单生产者单消费者SPSC环形队列以及教学版的 Treiber 栈。Treiber 栈会暴露 ABA 问题和内存回收问题这两点恰恰是随手写一个无锁结构最容易踩空的地方。文中所有代码以 C17 为基准只使用标准库本机没有 C 工具链代码未经编译正确性依据是标准语义与逐行推演。需要提前说明标准库并不保证std::atomicT对任意T都是无锁的。标准只要求std::atomic_flag是无锁的对其他特化只要求应当无锁具体要用is_lock_free()查询。C17 增加了is_always_lock_free静态成员可以在编译期判断某个特化是否在所有平台上都无锁。把用了atomic直接等同于无锁算法是第一个要纠正的误解。一、进度保证lock-free、wait-free 与 obstruction-free这三个词都有形式化定义日常讨论中经常被混用。粗略地说obstruction-free无阻塞只要其他线程都停下来某个线程能在有限步内完成操作。这是最弱的保证一个用 CAS 重试的循环通常至少满足它。lock-free无锁整个系统始终有进展即任意时刻至少有一个线程能在有限步内完成自己的操作。某个线程可能一直失败饥饿但系统不会整体卡死。wait-free无等待每个线程都能在有限步内完成不依赖其他线程的进度。这是最强的保证实现代价通常也最高。对比互斥量持锁线程如果被操作系统挂起其他线程全部阻塞系统没有任何线程能推进因此互斥量连 obstruction-free 都不满足。这就是无锁结构在线程被抢占信号处理这类场景下的价值所在。但要注意lock-free 不等于没有原子操作也不等于没有循环等待。CAS 重试循环本身就是一种等待只是它不会把线程交给操作系统去睡眠。另外系统有进展并不保证你的那次push能成功——它可能一直重试。二、CAS 与内存序无锁结构的两块地基几乎所有无锁结构都建立在读—比较—交换之上。标准库提供的接口是compare_exchange_weak与compare_exchange_strong签名大致如下以std::atomicT为例bool compare_exchange_weak(T expected, T desired,std::memory_order success,std::memory_order failure) noexcept;语义是若当前值与expected相等则写入desired并返回true否则把当前值写回expected并返回false。weak版本允许伪失败spurious failure即使值相等也可能返回false所以必须放在循环里strong版本不会伪失败但在某些平台上可能生成更贵的指令。无锁循环的惯用写法是用weak版本。第二个地基是内存序。默认的memory_order_seq_cst最容易推理代价也最高。常用的优化组合是生产者写完数据后用release存储发布数据已就绪消费者用acquire读取同一个原子变量看到发布值后之前写的数据对它可见。release与acquire配对建立 happens-before 关系relaxed只保证该原子变量自身的原子性和修改顺序不建立任何跨变量的同步。用relaxed传递指针或下标是数据竞争和读到半初始化对象的常见来源。三、ABA 问题与安全的内存回收先看教学版 Treiber 栈的核心逻辑// 适用于 C17仅用于说明 ABA不能直接用于多线程释放节点#include atomictemplate typename Tclass TreiberStack {struct Node {T value;Node* next;};std::atomicNode* head_{nullptr};public:void push(const T v) {Node* n new Node{v, head_.load(std::memory_order_relaxed)};while (!head_.compare_exchange_weak(n-next, n,std::memory_order_release,std::memory_order_relaxed)) {// 失败时 n-next 已被更新为当前 head继续重试}}bool pop(T out) {Node* old head_.load(std::memory_order_acquire);while (old ! nullptr !head_.compare_exchange_weak(old, old-next,std::memory_order_acquire,std::memory_order_relaxed)) {// 失败时 old 被更新为最新 head}if (old nullptr) {return false;}out old-value;delete old; // 危险别的线程可能刚读到 old正在读 old-nextreturn true;}};这段代码的push在单生产者下是可行的但pop有两个致命问题。第一是ABA。线程 A 读到head a、a-next b随后被挂起线程 B 依次弹出a、弹出b又把a重新压回栈顶。此时 A 恢复执行CAS 发现head仍然是a于是成功地把head改成b——而b可能已经被释放。CAS 只比较值不比较这个值期间有没有变过。第二是释放后使用use-after-free。即使没有 ABAA 在 CAS 成功后delete old另一个线程可能仍持有指向old的指针。工程上的解决方向有三类带标签的指针把版本号与指针打包进一个足够宽的类型用双字 CAS 一起更新。x86-64 上有 16 字节的cmpxchg16b但标准 C 不保证存在 128 位无锁原子类型可移植性差。安全内存回收hazard pointer每个线程登记自己正在读的节点、epoch/RCU延迟到所有读者退出临界区后再释放、引用计数。这些都需要额外的协议不是改几行代码就能完成。不做手动回收用std::shared_ptr管理节点。C20 起有std::atomicstd::shared_ptrT特化但标准并不保证它无锁多数实现内部仍用锁或引用计数表需要 C20 及以上使用前应查实现的is_lock_free()。结论是一个正确的无锁栈难点不在 CAS 循环而在内存回收。如果只是需要一个并发栈std::mutex加std::vector往往更简单也更不容易错。四、SPSC 环形队列约束明确时才可靠单生产者单消费者是约束最强、也最容易写对的场景。因为两端各只有一个线程不需要 CAS只需要用 acquire/release 传递下标并遵守容量留一格的约定区分空与满。// 适用于 C17#include atomic#include cassert#include cstddef#include vectortemplate typename Tclass SpscRingBuffer {public:explicit SpscRingBuffer(std::size_t capacity): buf_(capacity), cap_(capacity), head_(0), tail_(0) {assert(capacity 2);}// 仅生产者线程调用bool push(const T value) {const std::size_t tail tail_.load(std::memory_order_relaxed);const std::size_t next (tail 1) % cap_;if (next head_.load(std::memory_order_acquire)) {return false; // 满留一格不用}buf_[tail] value;tail_.store(next, std::memory_order_release);return true;}// 仅消费者线程调用bool pop(T out) {const std::size_t head head_.load(std::memory_order_relaxed);if (head tail_.load(std::memory_order_acquire)) {return false; // 空}out buf_[head];head_.store((head 1) % cap_, std::memory_order_release);return true;}std::size_t capacity() const { return cap_ - 1; }private:std::vectorT buf_;std::size_t cap_;std::atomicstd::size_t head_;std::atomicstd::size_t tail_;};几个关键点生产者只写tail_消费者只写head_两个原子变量各自只有一个写者因此不需要 CAS普通的 load/store 就够。buf_[tail] value必须在tail_.store(release)之前完成消费者用acquire读到新的tail_后才能安全读buf_[head]。这条 release/acquire 链是整个队列正确性的核心。容量留一格使head tail表示空、(tail 1) % cap head表示满避免额外维护计数。这个实现要求T可默认构造std::vectorT buf_(capacity)会值初始化元素且可拷贝赋值。若要支持只可移动的类型需要改用未初始化存储加 placement new复杂度和异常安全都会上升。alignas(64)常被用来把head_和tail_分到不同缓存行减少伪共享false sharing。但 64 只是常见 x86 平台的缓存行大小标准不保证而且是否真的分到不同行还取决于对象布局。五、什么时候不该用无锁无锁结构适合临界区极短、竞争可预测、对线程被挂起敏感如实时系统、信号处理、或者已经有成熟库实现如 Boost.Lockfree的场景。不适合临界区里要分配内存、要调用可能抛异常的函数、逻辑复杂到需要多个原子变量保持一致。多个原子变量无法用一个 CAS 原子地更新这时要么引入版本号把状态打包要么退回互斥量。常见坑点❌ 认为用了std::atomic就是无锁算法。✅ 先查is_lock_free()std::atomicT对大对象可能内部加锁算法层面也可能仍需 CAS 重试。❌ 用memory_order_relaxed发布指针或下标再让另一个线程解引用。✅ 发布用release获取用acquire不确定时直接用默认的seq_cst。❌pop拿到节点后立刻delete。✅ 用 hazard pointer、epoch 或引用计数延迟回收否则就是释放后使用。❌ 用普通bool做停止标志写成while (!stop) {}。✅ 用std::atomicbool并以acquire读取普通变量的并发读写是数据竞争属于未定义行为。❌ 把 SPSC 队列直接给多个生产者使用。✅ 多生产者需要 CAS 抢占槽位或改用 MPSC 设计单写者假设被破坏后tail_的普通 store 会丢失更新。❌ 用volatile代替原子变量做线程间同步。✅volatile不提供原子性与内存序只适合内存映射 I/O 等特殊场景线程同步用std::atomic或互斥量。❌ 断言无锁一定比std::mutex快。✅ 竞争激烈时 CAS 重试会浪费 CPU应当用std::chrono::steady_clock或基准测试框架在自己的负载上测量并说明测量方法与噪声来源。❌ 在无锁结构中让多个原子变量分别更新期望它们同时生效。✅ 单次 CAS 只能原子地更新一个对象需要一致性的多字段状态应打包进一个结构并保证其无锁或改用锁。总结概念保证典型代价wait-free每个线程有限步完成实现最复杂常需额外空间lock-free系统整体始终有进展CAS 重试、ABA、内存回收obstruction-free其他线程停止时可完成单独运行才有保证mutex无进度保证但语义简单阻塞、优先级反转、死锁风险无锁的核心不是删掉mutex而是用原子操作和内存序重新建立线程间的同步关系并为节点回收设计一套协议。真正难的是 ABA 与释放后使用而不是 CAS 循环本身。对大多数业务代码先用互斥量把逻辑写对只有在测量证明锁是瓶颈、且场景满足单写者或可安全回收等约束时再考虑无锁实现。

C++无锁数据结构实现示例详解

相关推荐

【算法学习笔记】LeetCode 15：三数之和（排序 + 双指针）

五金工具小批量多批次，网易外贸通怎么接住

降ai免费入口教程：比话上传论文先免费看预估AI率，再免费降AI 500字

更多资讯

PySceneDetect视频场景智能检测：从零开始的完整指南

palera1n 越狱完整手把手教程：从编译到还原

PySceneDetect场景检测实战指南：5种核心算法与应用技巧

rea 快速指南：三条命令逆向你的第一个二进制，每个结论带证据

dsh-market 快速上手：DSH 插件市场 3 步装好，主题一键安装免重启

编译过程理解

今日更新

多智能体协作系统架构设计与实践优化

大模型系统性入门实战指南：从原理到部署微调的全链路路线

鸿蒙文件系统沙箱机制解析与开发实践

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？