cpp无锁环形队列（MPMCc&WorkSteal queue）的实现

2026.09.05

cpp是一门不得不自己造轮子的语言，在这其中并发安全的无锁队列无疑是开发者常常面对的棘手的难题。在面临特定的低延迟场景中如果确定不会发生队列溢出或者明确需要避免malloc动态扩展队列长度时，往往会使用环形队列。本文罗列cpp内存模型下无锁环形队列的实现，及不同场景下对应的不同，让读者了解到手写实现无锁环形队列中那些可能会踩的坑。

在开始前先约定全文的游标定义：

head_：入队端 / 工作端（push 写入端，向前推进 head_++）；
tail_：出队端 / 消费端（pop 读出端，向前追赶 tail_++）；
队列中的有效数据区间为 [tail_, head_)，队列中元素数量为 head_ - tail_。

很容易想到的最简雏形，是通过 CAS 抢占 head_ / tail_ 来分配槽位：

class Queue {
  std::atomic<size_t> head_{0};
  std::atomic<size_t> tail_{0};
  std::array<T, Size> buffer_;

  bool push(T item) {
    size_t head = head_.load(std::memory_order_relaxed);
    size_t tail = tail_.load(std::memory_order_acquire);
    if (head - tail < Size && head_.compare_exchange_weak(head, head + 1, std::memory_order_relaxed)) {
      buffer_[head % Size] = std::move(item);
      return true;
    }
    return false;
  }

  bool pop(T& item) {
    size_t tail = tail_.load(std::memory_order_relaxed);
    size_t head = head_.load(std::memory_order_acquire);
    if (tail < head && tail_.compare_exchange_weak(tail, tail + 1, std::memory_order_relaxed)) {
      item = std::move(buffer_[tail % Size]);
      return true;
    }
    return false;
  }
};

初看似乎通过比较 head - tail 解决了套圈问题，但多线程并发下这里有致命缺陷：CAS 只是抢到了槽位的下标，并不代表槽位中的数据已经准备好了。

并发环境下 buffer 槽位的读写是一个三状态过程：

写入端不是瞬间完成的：它必然经历 未写（空闲） $\rightarrow$ 正在写（占用中） $\rightarrow$ 已写入（数据就绪）；
读出端同样如此：经历 未读（就绪） $\rightarrow$ 正在读（占用中） $\rightarrow$ 已读出（空闲）。

为了表达这三个状态、同时抵御线程挂起导致的“套圈”，必须引入版本号（sequence）。 这正是 Dmitry Vyukov 经典 MPMC 环形队列的核心思想：为每个槽位维护一个单调推进的序列号，将槽位的状态转移与全局的入队/出队游标严格绑定。

class Queue {
  struct cell_t
  {
    std::atomic<size_t>   sequence_;
    T                     data_;
  };
  std::atomic<size_t> head_{0};
  std::atomic<size_t> tail_{0};
  std::array<cell_t, Size> buffer_;

  bool push(T const& data)
  {
    cell_t* cell;
    size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
    for (;;)
    {
      cell = &buffer_[pos % Size];
      size_t seq = 
        cell->sequence_.load(std::memory_order_acquire);
      // intptr_t 和 std::size_t 有相同位宽
      intptr_t dif = (intptr_t)seq - (intptr_t)pos;
      if (dif == 0)
      {
        if (enqueue_pos_.compare_exchange_weak
            (pos, pos + 1, std::memory_order_relaxed))
          break;
      }
      //队列已满
      else if (dif < 0)
        return false;
      // 其他的producer和consumer已经转了一圈
      else
        pos = enqueue_pos_.load(std::memory_order_relaxed);
    }
    cell->data_ = data;
    // +1 代表数据有效
    cell->sequence_.store(pos + 1, std::memory_order_release);
    return true;
  }

  bool pop(T& data)
  {
    cell_t* cell;
    size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
    for (;;)
    {
      cell = &buffer_[pos % Size];
      size_t seq = 
        cell->sequence_.load(std::memory_order_acquire);
      intptr_t dif = (intptr_t)seq - (intptr_t)(pos + 1);
      if (dif == 0)
      {
        if (dequeue_pos_.compare_exchange_weak
            (pos, pos + 1, std::memory_order_relaxed))
          break;
      }
      else if (dif < 0)
        return false;
      else
        pos = dequeue_pos_.load(std::memory_order_relaxed);
    }
    data = cell->data_;
    cell->sequence_.store
      (pos + Size, std::memory_order_release);
    return true;
  }

}

mpsc 下的变化

有些场景下，业务逻辑能保证带个consumer，即pop操作只在一个thread上被调用，在上文的mpmc的基础上能够进一步优化

  bool Pop(T& data) {
    cell_t* cell = &buffer_[dequeue_pos_ & mask_];
    std::size_t seq = cell->sequence.load(std::memory_order::acquire);
    auto diff = static_cast<std::ptrdiff_t>(seq) - static_cast<std::ptrdiff_t>(dequeue_pos_ + 1);
    if (diff < 0) [[unlikely]] {
      return false;  // queue empty
    }
    // diff >= 0 的分支不会触发
    data = std::move(cell->data);
    cell->sequence.store(dequeue_pos_ + capacity_, std::memory_order_lelaxed);
    ++dequeue_pos_;
    return true;
  }

WorkSteal queue

worksteal 场景下通常使用SPMC队列，这里面有两个基本条件很重要：

Owner线程的push和pop都发生在head端，窃取者线程（Thief Threads）只会发生tail端的pop（steal）
worksteal 线程池中往往存放的是一个任务句柄（32/64位指针），所以可以将buffer特化为 std::atomic<T> _buffer[Size]; 保障读写的无锁原子性，当然代价是 T 只能存整数或裸指针。（现代硬件体系结构（如 x86-64、ARM64）的原子指令一般只支持 1-8 字节，部分架构带扩展指令（如 x86 的 CMPXCHG16B 或 ARM 的 LDXP/STXP）最多支持 16 字节，当然对任意类型的T，std::atomic<T> 的load/store操作都是具有原子性保障的，但是编译器是通过加入互斥锁来实现）

在以上背景下，SPMC的buffer读写其实退化为了两个状态（未读/已读和未写/已写），回想MPMC的读写是三个状态（未读/在读/已读），笔者认为可以将MPMC的版本号（sequence）理解为是为了处理在读和在写的状态矛盾另一部分是为了解决多线程MPMC下某个线程被挂起，而槽位被下一轮复用，所以如果同时解决读写的原子性和producer 或 consumer 任意一方是单线程的，即可不使用版本号就能保证并发安全。很容易就能想到以下程序

// try_push 端(producer/owner)：
_head = head_.load(std::memory_order_relaxed);
_tail = tail_.load(std::memory_order_acquire);
if (_head < _tail + Size - 1)
  buffer[_head & BufferMask].store(..., std::memory_order_relaxed); //  写入数据
  head_.store(_head + 1, std::memory_order_release);              //  发布新边界

// try_pop 端(producer/owner)：
_head = head_.load(std::memory_order_relaxed);
_tail = tail_.load(std::memory_order_acquire);
if (_head > _tail)
// 如果此时有其他线程执行steal不保证后面 head > tail
  ... = buffer[(_head - 1) & BufferMask].load(std::memory_order_relaxed); // 读取数据
  head_.store(_head - 1, std::memory_order_release);             //  发布新边界

// steal 端(thief)：
_tail = tail_.load(std::memory_order_acquire);
_head = head_.load(std::memory_order_acquire);                   //  获取同步
if (_head > _tail)
  ... = buffer[_tail & BufferMask].load(std::memory_order_relaxed); //  读取数据
  if (!tail_.compare_exchange_strong(_tail, _tail + 1, std::memory_order_seq_cst, std::memory_order_relaxed))
    // CAS 抢到则成功偷取，CAS抢不到则返回失败

但是这样又产生了新的问题，cpu存在Store-Load 重排， pop 遇上其他线程上的steal如果发生重排会发生潜在的冲突： 写未刷出，读已发生：在 pop 中，Owner 需要先将自己的边界往回缩（写入新边界 head），然后再读取 Thief 的位置 tail 确认队列是否还有剩余元素。而在 steal 端，Thief 也是先读取 tail 再读取 Owner 的边界 head。如果 CPU 发生了 Store-Load 重排，Owner 对新边界 head 的写入尚未刷新到总线对其他核可见，Owner 就已经提前读取了 Thief 的位置 tail；与此同时，Thief 也读到了 Owner 尚未缩小的旧 head。这会导致双方都判定“该槽位属于自己”，最终将同一个任务分别取出执行，导致并发重复消费！

好直接看一下代码：

class BoundedWSQ {
  static constexpr size_t BufferSize = 1024;
  static constexpr size_t BufferMask = BufferSize - 1;

  alignas(64) std::atomic<int64_t> head_{0}; // 工作端 (Owner 负责 push 和 pop)
  alignas(64) std::atomic<int64_t> tail_{0}; // 窃取端 (Thief 负责 steal)
              int64_t              cached_tail_{0};
  alignas(64) std::atomic<T>       buffer_[BufferSize];

public:
  bool try_push(T item) {
    int64_t h = head_.load(std::memory_order_relaxed);

    // 乐观检查：先用本地缓存的 cached_tail_，避免跨核缓存失效开销
    if (h - cached_tail_ >= BufferSize) {
      cached_tail_ = tail_.load(std::memory_order_acquire);
      if (h - cached_tail_ >= BufferSize) {
        return false; // 队列满
      }
    }

    buffer_[h & BufferMask].store(item, std::memory_order_relaxed);
    head_.store(h + 1, std::memory_order_release);
    return true;
  }

  T pop() {
    int64_t h = head_.load(std::memory_order_relaxed) - 1;
    head_.store(h, std::memory_order_relaxed);
    std::atomic_thread_fence(std::memory_order_seq_cst); // 阻断 Store-Load 重排
    int64_t t = tail_.load(std::memory_order_relaxed);

    if (t <= h) {
      T item = buffer_[h & BufferMask].load(std::memory_order_relaxed);
      if (t == h) {
        // 队列只剩最后一个元素，与可能的 Thief 竞争 tail
        if (!tail_.compare_exchange_strong(t, t + 1,
              std::memory_order_seq_cst, std::memory_order_relaxed)) {
          item = nullptr; // 已经被 Thief 偷走
        }
        head_.store(h + 1, std::memory_order_relaxed);
      }
      return item;
    }

    head_.store(h + 1, std::memory_order_relaxed); // 队列为空，恢复 head
    return nullptr;
  }

  T steal() {
    int64_t t = tail_.load(std::memory_order_acquire);
    std::atomic_thread_fence(std::memory_order_seq_cst);
    int64_t h = head_.load(std::memory_order_acquire);

    if (t < h) {
      T item = buffer_[t & BufferMask].load(std::memory_order_relaxed);
      if (!tail_.compare_exchange_strong(t, t + 1,
            std::memory_order_seq_cst, std::memory_order_relaxed)) {
        return nullptr; // CAS 抢输，返回失败
      }
      return item;
    }

    return nullptr;
  }
};

pop 和 steal 并发问题

WSQ 最最绕的地方，就是 Owner 的 pop 和 Thief 的 steal 在只剩最后一个元素时的争抢的。

先减 head_ 占坑：有效区间是 [tail, head)。如果 Owner 先去读 tail_ 看有没有元素再减 head_，万一读完瞬间线程被切走，这期间 thief 们把任务全偷光了，Owner 醒来再减 head_ 拿数据就会拿到脏数据。所以 Owner 得先把 head_ 减 1 缩减范围，告诉后面的小偷别碰这个位置。
steal 是分步的（先读 tail，再读 head 确认非空，最后 CAS 推进 tail）。在 Owner 减 head_ 前的一瞬间，可能已经有某个 thief 读到了旧的 head，正准备执行 CAS。如果 Owner 发现 t == h（只剩最后一份数据）还不去抢 tail，steal 紧接着 CAS 也成功，这个任务就会被两边各执行一次，所以 Owner 去 CAS 抢 tail_ + 1，就是为了把小偷手里的 CAS 预期值破坏掉。抢赢了独占任务，抢输了说明小偷已经先得手了，Owner 认输把 head_ 加回去就行。
内存屏障：在写完 head_ 和读 tail_ 之间必须插一条 seq_cst fence，防止硬件把 Store-Load 乱序，确保两边的最新状态能及时对齐。

总结

MPMC 队列：本质上是用每个槽位的 sequence（版本号） 来充当细粒度的状态机，配合 CAS 来保证没有并发冲突与套圈问题。Vyukov 的 MPMC Queue 逻辑较规则，完全可以自己手写实现并应用到通用线程池中。
Work-Stealing Queue（Chase-Lev WSQ）：则是借助 std::atomic<T> 本身的原子读写（要求存指针或整数等满足平凡拷贝的类型），配合各个特定位置的内存屏障（如 seq_cst fence 解决 Store-Load 乱序）来保障并发安全。它最大的复杂度就在于 Owner 本地 pop 和 Thief 并发 steal 在单元素时的冲突问题，通过让 pop 先减 head_ 后 CAS 抢 tail_ 解决。对于 Work-Stealing 这种涉及双端操作和复杂边界的队列，细节很多，工程实践中还是尽量参考成熟的工业级实现。

参考实现：

Dmitry Vyukov MPMC Bounded Queue: https://github.com/craflin/LockFreeQueue/blob/master/mpmc_bounded_queue.h
Taskflow Work-Stealing Queue: https://github.com/taskflow/work-stealing-queue/blob/master/wsq.hpp