前言


生产者—消费者模型里，队列是线程之间最常见的交接点。「有锁队列」和「无锁队列」经常被拿来比较，但很多介绍把两者讲成了「慢」和「快」的关系，这是不准确的：它们的差别主要在阻塞语义、内存序约束和内存回收代价，而不是一条绝对的性能排序。脱离场景谈谁更快没有意义——在争用激烈时，一个设计良好的互斥量队列完全可能跑赢一个写得不讲究的无锁队列。无锁真正换来的性质是：某个线程被操作系统换出调度时，不会把其他线程一起拖住。


另一个必须纠正的误解是把 volatile 当成线程同步手段。volatile 只告诉编译器「不要把这个访问优化掉」，它不提供原子性，也不建立 happens-before 关系，用它做线程间通信是错的：读到的可能是撕裂的值，编译器和 CPU 也不保证写入对其他核心可见的顺序。C++ 里做无锁编程的正确工具是 std::atomic 配上一组显式内存序。


本文先用「互斥量 + 条件变量」写一个有锁队列，再给出一个单生产者单消费者（SPSC，single producer single consumer）的无锁环形队列完整实现，然后讲清 ABA 与内存序这两个真正的陷阱，最后给出选型对照。所有代码以 C++17 为基准，在 GCC 13 / Clang 17 / MSVC 19.3x 上均可编译。


一、有锁队列：互斥量 + 条件变量


有锁队列的核心是把「判空 / 判满」「入队 / 出队」这些复合操作用一把互斥量（mutex）圈起来。互斥量的 lock() 隐含 acquire 语义，unlock() 隐含 release 语义，因此临界区内的所有写在解锁之后对其他线程可见——这是标准规定，不是实现细节。换句话说，用对了锁，就不需要再关心内存序。


下面这段是可以直接编译运行的完整实现。它比常见的教学代码多了一个 close()：这是为了让消费者能从阻塞等待中退出来，否则析构时消费者会永远卡在 wait_pop 里。


// locked_queue.cpp  编译：g++ -std=c++17 -pthread locked_queue.cpp -o locked_queue

#include <condition_variable>

#include <iostream>

#include <mutex>

#include <queue>

#include <thread>

#include <utility>



template <typename T>

class LockedQueue {

public:

    // 入队：拷贝/移动进队列，然后唤醒一个等待者

    void push(T value) {

        {

            std::lock_guard<std::mutex> lock(mutex_);

            queue_.push(std::move(value));

        }   // 锁在这里释放

        cond_.notify_one();

    }



    // 非阻塞出队：队列空就立刻返回 false

    bool try_pop(T& out) {

        std::lock_guard<std::mutex> lock(mutex_);

        if (queue_.empty()) {

            return false;

        }

        out = std::move(queue_.front());

        queue_.pop();

        return true;

    }



    // 阻塞出队：队列空就睡，直到有元素或被 close()

    bool wait_pop(T& out) {

        std::unique_lock<std::mutex> lock(mutex_);

        // 谓词形式：wait 内部会循环检查，天然免疫虚假唤醒

        cond_.wait(lock, [this] { return closed_ || !queue_.empty(); });

        if (queue_.empty()) {

            return false;   // 已关闭且已排空

        }

        out = std::move(queue_.front());

        queue_.pop();

        return true;

    }



    // 关闭：唤醒所有等待者，让它们从 wait_pop 返回

    void close() {

        {

            std::lock_guard<std::mutex> lock(mutex_);

            closed_ = true;

        }

        cond_.notify_all();

    }



private:

    std::queue<T> queue_;

    std::mutex mutex_;

    std::condition_variable cond_;

    bool closed_ = false;

};



int main() {

    LockedQueue<int> q;



    std::thread producer([&q] {

        for (int i = 0; i < 100; ++i) {

            q.push(i);

        }

        q.close();

    });



    int sum = 0;

    int value = 0;

    while (q.wait_pop(value)) {

        sum += value;

    }

    producer.join();

    // 输出固定为 4950（0+1+...+99），与线程调度无关

    std::cout << "sum = " << sum << '\n';

    return 0;

}

几个值得注意的细节：



  • notify_one() 放在解锁之后。持锁通知不会出错，但被唤醒的线程会立刻又因为拿不到锁而重新阻塞，多一次无谓的上下文切换。这是性能考量，不是正确性要求。

  • wait(lock, pred) 的谓词在持锁状态下求值（wait 内部会先重新加锁再检查），所以谓词里读 closed_ 和 queue_ 是安全的。

  • out = std::move(queue_.front()) 之后再 pop()：std::queue::front() 返回的是元素的引用，把它的内容移走之后再弹出节点是正确的；不要先把 front() 的引用存下来再 pop()，那样会得到悬垂引用。


二、无锁队列：原子操作 + 环形缓冲


无锁指针式队列（比如 Michael–Scott 队列）最难的部分不是入队出队本身，而是内存回收：你无法知道某个被 pop 掉的节点还有没有别的线程正拿它做 CAS，所以不能直接 delete。这需要危险指针（hazard pointer）、引用计数或基于 epoch 的回收方案，代码量和出错面都很大。


因此这里给出的可编译实现是定长环形缓冲（ring buffer）版本的 SPSC 队列。它把「回收问题」变成了「覆写问题」：槽位是固定数组，索引只向前走，不存在释放时机的问题，也就不存在 ABA。代价是只支持一个生产者和一个消费者，且容量固定。


// spsc_ring.cpp  编译：g++ -std=c++17 -pthread spsc_ring.cpp -o spsc_ring

#include <array>

#include <atomic>

#include <cstddef>

#include <iostream>

#include <thread>



// 64 是 x86-64 上常见的缓存行宽度。这是体系结构相关的事实，不是标准规定：

// 某些 ARM 平台的缓存行是 128 字节，移植时以目标平台为准。

constexpr std::size_t kCacheLine = 64;



// 约束：T 需要可默认构造（用于 buffer_ 的初始化）与可拷贝赋值（用于读写槽位），

// 这与 std::vector 之类的容器一样，是可复制的元素类型的基本要求。

template <typename T, std::size_t Capacity>

class SpscRing {

    static_assert(Capacity >= 2, "Capacity 至少要能放下一个元素");

    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity 必须是 2 的幂");



public:

    // 只允许生产者线程调用

    bool push(const T& value) {

        const std::size_t head = head_.load(std::memory_order_relaxed);

        const std::size_t next = (head + 1) & (Capacity - 1);

        if (next == tail_.load(std::memory_order_acquire)) {

            return false;                    // 满：留一个空槽区分满和空

        }

        buffer_[head] = value;               // 先写数据

        head_.store(next, std::memory_order_release);   // 再发布索引

        return true;

    }



    // 只允许消费者线程调用

    bool pop(T& out) {

        const std::size_t tail = tail_.load(std::memory_order_relaxed);

        if (tail == head_.load(std::memory_order_acquire)) {

            return false;                    // 空

        }

        out = buffer_[tail];                 // 先读数据

        tail_.store((tail + 1) & (Capacity - 1), std::memory_order_release);

        return true;

    }



private:

    std::array<T, Capacity> buffer_{};

    alignas(kCacheLine) std::atomic<std::size_t> head_{0};

    alignas(kCacheLine) std::atomic<std::size_t> tail_{0};

};



int main() {

    constexpr int kCount = 1000;

    SpscRing<int, 8> ring;   // 容量 8，实际可用 7 个槽



    std::thread producer([&ring] {

        for (int i = 1; i <= kCount; ++i) {

            while (!ring.push(i)) {

                std::this_thread::yield();   // 满了就礼让，避免纯自旋

            }

        }

    });



    long long sum = 0;

    int value = 0;

    int received = 0;

    while (received < kCount) {

        if (ring.pop(value)) {

            sum += value;

            ++received;

        } else {

            std::this_thread::yield();

        }

    }

    producer.join();

    // 输出固定为 500500（1+2+...+1000）

    std::cout << "sum = " << sum << '\n';

    return 0;

}

为什么是这两组内存序


head_ 只被生产者写、消费者读；tail_ 只被消费者写、生产者读。这就是 SPSC 的全部约束。



  • 读自己的索引用 relaxed：head_ 在 push 里只由本线程改，不存在与别人的竞争，读一个「大致最新」的值就够了。

  • 读对方的索引用 acquire：这是关键。生产者读到 tail_ 的 release 写入后，才能保证消费者在更新 tail_ 之前对 buffer_ 槽位的读取已完成，从而不会覆写一个还没被取走的元素。反之，消费者用 acquire 读 head_，才能保证它读到的 buffer_[tail] 是生产者写完整的数据。

  • 写自己的索引用 release：保证「写数据」排在「发布索引」之前，不会被编译器或 CPU 重排到后面去。


如果把 acquire / release 全换成 relaxed，程序在 x86 上可能看起来正常工作（x86 的 TSO 内存模型对 store-load 之外的重排比较宽容），但这是数据竞争加内存序错误，属于未定义行为，标准不保证任何行为，换到 ARM 或换一个编译器版本就可能读到半写状态。


顺带一提：std::atomic<std::size_t> 在主流平台上是无锁的，但标准并不强制。如果这段代码要在不常见的目标平台上跑，可以用 std::atomic<std::size_t>::is_always_lock_free 在编译期检查一下。


三、ABA 与内存序陷阱


ABA 问题是 CAS（compare-and-swap）循环特有的。考虑指针式队列里的出队操作：线程 T1 读到头指针 p，算出 next = p->next，然后准备 CAS(head, p, next)。在它执行 CAS 之前，线程 T2 把 p 弹走、p 被回收、同一块内存又被重新分配给了新的节点，并被压回队列头。此时 T1 的 CAS 看到的仍然是地址 p，它会误以为头指针没变，于是把 head 改成那个早已失效的 next——队列结构就此损坏。


ABA 不是「概率很低的偶发 bug」，它是结构性的：只要内存可以被回收并复用，它就存在。工程上的主流做法是：



  2. 打标签：把「指针 + 版本号」打包成一个整体做 CAS。C++17 标准里没有可移植的双字宽 CAS（std::atomic 对大于指针宽度的类型通常退化成加锁实现，is_lock_free() 返回 false），所以这一招通常要靠编译器内建函数（如 GCC 的 __atomic_compare_exchange_16）或平台 intrinsic。

  4. 延迟回收：危险指针、epoch 回收、std::shared_ptr 的原子特化——让节点在被真正释放前，保证没有线程还持有它。

  6. 回避：换成定长环形缓冲，压根不分配释放节点，问题消失。本文第二节就是这么做的。


内存序陷阱方面，最常见的是「用 seq_cst 包打天下」和「用 relaxed 省事」两个极端。seq_cst 单独用是正确的（只是可能比需要的约束更强），relaxed 用在需要建立顺序的地方则是错误的。判断标准很简单：只要一个线程写了数据、另一个线程要读它，两者之间就必须有一对 release/acquire（或者间接的关系，比如通过 seq_cst 的全序、或通过 atomic_thread_fence）才能建立 happens-before。


最后再强调一次：不要让 volatile 出现在并发代码里做同步。它唯一可靠的用途是 volatile 语义本身——比如访问映射到硬件的内存、或者在 setjmp/longjmp 之间保护局部变量。


常见坑点




场景❌ 错误写法✅ 正确写法



条件变量等待cond_.wait(lock); 裸等，虚假唤醒后会读到空队列`cond_.wait(lock, [this]{ return closed_ \

volatile 同步volatile bool flag; while (!flag) {}std::atomic<bool> flag; while (!flag.load(std::memory_order_acquire)) {}

无锁代码内存序全用 memory_order_relaxed，靠"看起来能跑"发布索引用 release，读对方索引用 acquire

无锁队列退出消费者只检查「队列是否空」就退出用固定条数或独立的原子结束标志，避免丢失最后几个元素

有锁队列析构消费者卡在 wait_pop，析构函数里直接销毁对象先 close() 唤醒所有等待者，join() 后再析构

std::thread 生命周期线程对象析构时仍 joinable()析构前 join() 或 detach()，否则调用 std::terminate

伪共享两个索引紧挨着放，每次更新都来回弹缓存行用 alignas(64) 把索引分开到不同缓存行

搬起石头砸脚把 front() 的引用存下来再 pop()先取值/移动再 pop()，或用 std::move(q.front()) 后立即 pop()



有一处需要格外留意：无锁队列的「生产者已完成」通知和「队列已排空」是两个独立事件。上面表格里那条「无锁队列退出」说的就是它——如果消费者先看到「队列为空」再看到「生产结束」，中间可能正好漏掉一个刚入队的元素。最稳的做法是按已知的总条数循环（本文示例就是数到 1000 才退出），或者在生产者的结束标志之前用一次 release 写、消费者用 acquire 读，并保证读到标志后再排空一次。


总结




维度有锁队列无锁 SPSC 环形队列



阻塞语义可以挂起等待，不浪费 CPU只能自旋重试或返回失败

生产者/消费者数量任意多严格一个生产者、一个消费者

内存回收锁天然保护，无需特殊处理定长数组，无回收问题

ABA 风险不存在环形索引不涉及；指针式队列才需要防范

内存序由 mutex 隐式保证需要手写 release/acquire

线程被换出时持锁线程被换出会拖住其他线程互不影响（这是无锁最大的收益）

出错后的表现死锁、忘记解锁，容易定位数据竞争，属于 UB，往往只在高负载或换平台后复现

实现与维护成本低高，且难以用调试器观察



选型建议很朴素：默认用有锁队列。它的正确性由语言和标准库替你保证，绝大多数业务场景的瓶颈也不在锁上。只有在测量确认互斥量是热点、且能明确限定为 SPSC 时，才考虑本文这种定长环形缓冲；如果确实需要多生产者多消费者的无锁结构，请优先使用经过充分验证的库，而不是自己从 CAS 开始写——那里的每一条内存序都需要能被论证，而不是靠「跑起来没崩」来判断。至于 volatile，无论在哪个方案里，它都不是同步原语。




# c++ # 开发语言
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

Web应用基础架构

该技术所提供的互操作性，推动了一致分布式体系架构的演进，该架构通常用于支持并简化那些复杂的分布式应用程序，它包括Web服务器，事务监控器和消息队列软件。简单来说是本身可视为电子化的文件柜，存储电子文件的处所，用户可以对文件中的数据进行新增、截取、更新、删除等操作，数据库拥有自己的语言，SQL查询语言。操作系统（Operating System，简称OS）是管理和控制计算机硬件与软件资源的计算机程序

avatar openEuler 社区
cover

C++条件变量condition_variable虚假唤醒到底是什么？为什么必须while循环判断？

avatar openEuler 社区

跨平台Node脚本怎样稳定解析Windows路径

在 Windows 上调试正常的文件名解析，放到另一种操作系统后可能把整段路径当成文件名。原因在于 node:path 的默认规则跟随运行平台，而输入数据可能来自另一套路径体系。离线解析清单时，规则应由数据格式决定，不能因为脚本恰好运行在某台机器上就悄悄切换。

avatar openEuler 社区