一、什么是 MPSC 队列前面实现线程安全队列时主要使用std::mutex保护共享队列。例如生产者添加数据std::lock_guardstd::mutex lock(_queueLock); _queue.push(value);消费者获取数据std::lock_guardstd::mutex lock(_queueLock); value _queue.front(); _queue.pop();这种方式很好理解但是每次操作共享队列都需要加锁 ↓ 访问队列 ↓ 解锁当多个线程频繁访问队列时就会出现多个线程竞争同一把锁的情况。于是还可以考虑另一种实现思路不使用传统的mutex保护整个队列而是利用std::atomic提供的原子操作完成线程之间的同步。这就会涉及无锁队列Lock-Free Queue。本篇先介绍其中非常典型的一种MPSC QueueMPSC 全称Multi Producer Single Consumer即Multi Producer 多个生产者 Single Consumer 单个消费者结构可以理解为生产者1 ───┐ │ 生产者2 ───┼──→ MPSC Queue ───→ 消费者 │ 生产者3 ───┘也就是说多个线程可以同时向队列写数据 只有一个线程负责从队列取数据例如日志系统中网络线程 ─────┐ │ 数据库线程 ───┼──→ 日志队列 ───→ 日志写入线程 │ 工作线程 ─────┘多个线程都会产生日志但是可以只安排一个线程真正将日志写入文件。这种场景就非常适合 MPSC 模型。和前面的生产者消费者队列相比mutex队列 生产者 ↓ mutex ↓ queue ↓ mutex ↓ 消费者MPSC 无锁队列则更倾向于生产者 ↓ atomic ↓ 链表节点 ↓ atomic ↓ 消费者这里的“无锁”主要指队列核心同步过程不依赖std::mutex而是使用std::atomic完成线程之间的同步。二、MPSC队列的核心结构Node、head和tail先来看队列内部最关键的数据结构templatetypename T class MPSCQueue { private: struct Node { T* Data; // 当前节点保存的数据 std::atomicNode* Next; // 指向下一个节点 Node() : Data(nullptr), Next(nullptr) {} explicit Node(T* data) : Data(data), Next(nullptr) {} }; std::atomicNode* _head; // 生产者操作的位置 std::atomicNode* _tail; // 消费者操作的位置 };和std::queue不同这里的队列底层实际上是一条单向链表Node ┌─────────────┐ │ Data │ │ Next ───────────→ └─────────────┘每一个节点包含两个东西T* Data;负责保存真正的数据。以及std::atomicNode* Next;保存下一个节点的地址。为什么Next不是普通的Node* Next;而是std::atomicNode* Next;因为生产者和消费者可能同时访问节点之间的连接关系。如果使用普通指针Node* Next;多个线程之间直接读写可能产生数据竞争。因此这里将指针本身声明为原子变量std::atomicNode* Next;整个链表大概是┌──────┐ ┌──────┐ ┌──────┐ │ Node │ ───→ │ Node │ ───→ │ Node │ ───→ nullptr │ 100 │ │ 200 │ │ 300 │ └──────┘ └──────┘ └──────┘除了节点以外还有两个重要指针std::atomicNode* _head; std::atomicNode* _tail;可以先简单理解成_head生产者主要操作的位置 _tail消费者主要操作的位置但这个实现初始化时不会直接让队列完全为空而是先创建一个哑节点Dummy NodeMPSCQueue() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed)) { Node* front _head.load(std::memory_order_relaxed); front-Next.store(nullptr, std::memory_order_relaxed); }刚创建队列时head ↓ ┌───────────┐ tail ─→ │ DummyNode │ ─→ nullptr └───────────┘也就是_head ↓ Dummy ↑ _tail这个 Dummy 节点并不保存真正需要处理的数据。它主要用于统一入队和出队操作减少对第一个元素 最后一个元素 空队列这些特殊情况的额外判断。例如真正加入数据以后Dummy → Node100 → Node200 → Node300Dummy 节点始终可以作为队列结构中的一个过渡节点。这种Dummy Node 哑节点 / 虚拟头节点在链表、队列以及很多无锁数据结构中都非常常见。三、Enqueue多个生产者如何同时添加数据MPSC 最关键的地方就是多个生产者如何在不使用 mutex 的情况下安全入队入队函数如下void Enqueue(T* input) { Node* node new Node(input); // 创建新节点 Node* prevHead _head.exchange(node, std::memory_order_acq_rel); // 原子交换head prevHead-Next.store(node, std::memory_order_release); // 将旧head连接到新节点 }代码只有三句但是这三句就是整个 MPSC 入队操作的核心。首先Node* node new Node(input);为新数据创建一个节点。例如queue.Enqueue(new Count(100));首先创建┌─────────────┐ │ Data 100 │ │ Nextnull │ └─────────────┘接下来Node* prevHead _head.exchange(node, std::memory_order_acq_rel);这里的exchange()是一个非常重要的原子操作。假设_head → Node100现在执行_head.exchange(Node200);它会完成两件事情第一步 把 _head 修改为 Node200 第二步 返回原来的 Node100所以Node* prevHead _head.exchange(node);执行之后prevHead → Node100 _head → Node200关键就在于exchange()整个过程是原子的。假设有两个生产者Producer1 Producer2同时执行_head.exchange(node);不会出现两个线程把_head修改乱掉的情况。原子变量会保证这些操作表现出一个确定的先后顺序。例如最开始_head → Dummy生产者1加入Node100执行Node* prevHead _head.exchange(Node100);得到prevHead → Dummy _head → Node100然后执行prevHead-Next.store(Node100);形成Dummy → Node100接下来生产者2添加Node200执行Node* prevHead _head.exchange(Node200);得到prevHead → Node100 _head → Node200然后prevHead-Next.store(Node200);最终Dummy → Node100 → Node200 ↑ _head如果继续添加 Node300Dummy → Node100 → Node200 → Node300 ↑ _head因此生产者每次实际上做的是创建新节点 ↓ 原子交换 _head ↓ 获得旧 head ↓ 旧 head 的 Next 指向新节点也就是Node* prevHead _head.exchange(node, std::memory_order_acq_rel); prevHead-Next.store(node, std::memory_order_release);这里为什么不用_head node;因为多个生产者可能同时修改_head。普通赋值并不能完成读取旧值 写入新值 返回旧值这一整套不可分割的操作。而exchange()可以原子地完成这一过程。这也是无锁代码中非常常见的写法。这里还涉及两个内存序std::memory_order_acq_rel std::memory_order_release可以先简单理解成exchange(..., acq_rel)既具有 acquire 的读取语义也具有 release 的发布语义。而Next.store(node, std::memory_order_release);表示生产者完成节点内容之后再把这个节点正式发布到链表中。这样消费者后面配合load(std::memory_order_acquire)读取时就能够正确看到生产者已经准备好的节点数据。所以生产者一侧可以概括为准备数据 ↓ release 发布 ↓ 消费者 acquire 获取四、Dequeue单消费者如何从队列中取出数据再来看消费者bool Dequeue(T* result) { Node* tail _tail.load(std::memory_order_relaxed); Node* next tail-Next.load(std::memory_order_acquire); if (!next) return false; result next-Data; _tail.store(next, std::memory_order_release); delete tail; return true; }由于 MPSC 的特点是多个 Producer 一个 Consumer所以真正执行Dequeue()的只有一个消费者线程。首先Node* tail _tail.load(std::memory_order_relaxed);获取当前_tail。假设现在_tail ↓ Dummy → Node100 → Node200那么tail指向Dummy然后Node* next tail-Next.load(std::memory_order_acquire);获取下一个节点。此时tail ↓ Dummy → Node100 ↑ next如果next nullptr说明当前没有可以取出的节点if (!next) return false;如果存在节点result next-Data;把数据返回给调用者。例如next-Data 100那么result → 100接着_tail.store(next, std::memory_order_release);让_tail向后移动。原来_tail ↓ Dummy → Node100 → Node200现在_tail ↓ Dummy → Node100 → Node200然后delete tail;删除原来的旧节点删除 Dummy最终_tail ↓ Node100 → Node200注意这里有一个比较容易第一次看懵的地方。执行result next-Data;取到的是next节点的数据但是delete tail;删除的却是上一个节点为什么不是直接delete next;这是因为当前的next还需要继续充当新的尾部节点。例如Dummy → Node100 → Node200第一次出队取出 Node100 的 Data 删除 Dummy _tail Node100变成Node100 → Node200 ↑ tail第二次出队取出 Node200 的 Data 删除 Node100 _tail Node200所以每次都是读取 next 的数据 删除旧 tail 让 next 成为新的 tail这其实也是 Dummy Node 设计带来的好处。整个出队流程可以总结成读取 tail ↓ 找到 tail-Next ↓ Next不存在 ↓ 是 队列暂时没有数据 ↓ 否 获取 next-Data ↓ tail移动到next ↓ 删除旧tail ↓ 返回true这里生产者发布节点时prevHead-Next.store(node, std::memory_order_release);消费者读取节点时tail-Next.load(std::memory_order_acquire);正好构成Producer release ↓ 节点数据发布 ↓ acquire Consumer这样消费者看到Next ! nullptr以后也能够正确看到这个节点之前已经准备好的Data这就是 acquire/release 在这个队列中的一个实际应用。五、完整MPSC队列与多线程测试把前面的结构组合起来可以得到一个简化后的 MPSC 非侵入式队列#ifndef MPSC_QUEUE_H #define MPSC_QUEUE_H #include atomic templatetypename T class MPSCQueue { private: struct Node { T* Data; std::atomicNode* Next; Node() : Data(nullptr), Next(nullptr) {} explicit Node(T* data) : Data(data), Next(nullptr) {} }; std::atomicNode* _head; std::atomicNode* _tail; public: // 创建Dummy节点head和tail最开始都指向它 MPSCQueue() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed)) {} // 禁止拷贝 MPSCQueue(const MPSCQueue) delete; MPSCQueue operator(const MPSCQueue) delete; ~MPSCQueue() { T* data; // 清理队列中还没有取出的数据 while (Dequeue(data)) delete data; // 删除最后留下的Dummy节点 Node* tail _tail.load(std::memory_order_relaxed); delete tail; } // 多个生产者可以调用Enqueue void Enqueue(T* input) { Node* node new Node(input); // 原子替换head并获得之前的head Node* prevHead _head.exchange(node, std::memory_order_acq_rel); // 将旧head连接到新节点 prevHead-Next.store(node, std::memory_order_release); } // 只允许一个消费者调用Dequeue bool Dequeue(T* result) { Node* tail _tail.load(std::memory_order_relaxed); // 获取tail后面的节点 Node* next tail-Next.load(std::memory_order_acquire); // 没有数据 if (!next) return false; // 获取数据 result next-Data; // tail向后移动 _tail.store(next, std::memory_order_release); // 删除旧tail delete tail; return true; } }; #endif测试时先定义一个简单的数据类型struct Count { explicit Count(int value) : v(value) {} int v; };然后创建MPSCQueueCount queue;两个生产者不断向队列添加数据一个消费者负责获取#include atomic #include iostream #include thread int main() { MPSCQueueCount queue; std::atomicint producerFinished{0}; // 生产者1 std::thread producer1([]() { queue.Enqueue(new Count(100)); queue.Enqueue(new Count(200)); queue.Enqueue(new Count(300)); queue.Enqueue(new Count(400)); producerFinished; }); // 生产者2 std::thread producer2([]() { queue.Enqueue(new Count(500)); queue.Enqueue(new Count(600)); queue.Enqueue(new Count(700)); queue.Enqueue(new Count(800)); producerFinished; }); // MPSC只有一个消费者 std::thread consumer([]() { Count* data nullptr; while (true) { if (queue.Dequeue(data)) { std::cout pop : data-v std::endl; delete data; continue; } // 两个生产者都结束并且当前已经取不到数据则结束消费 if (producerFinished.load() 2) break; std::this_thread::yield(); } }); producer1.join(); producer2.join(); consumer.join(); return 0; }这里线程关系非常清楚producer1 │ ├──100 ├──200 ├──300 └──400 │ ↓ ┌─────────┐ │ │ │ MPSC │ ─────→ consumer │ Queue │ │ │ └─────────┘ ↑ │ ┌──500 ├──600 ├──700 └──800 producer2和上一篇最大的区别是上一篇使用std::mutex std::condition_variable实现的是加锁 等待 唤醒这一篇则主要依靠std::atomicNode*以及exchange() load() store()维护链表。其中最关键的入队操作只有Node* prevHead _head.exchange(node, std::memory_order_acq_rel); prevHead-Next.store(node, std::memory_order_release);而消费者最关键的逻辑是Node* next tail-Next.load(std::memory_order_acquire);因此整个 MPSC 队列可以概括成多个生产者 ↓ atomic exchange ↓ 不断向链表尾部连接新节点 ↓ MPSC Queue ↓ 单个消费者 ↓ 沿着 Next 顺序取出数据学习这一部分最需要掌握的并不是一开始就把整个无锁队列背下来而是理解下面几个知识点之间的关系std::atomicNode* // 原子指针 exchange() // 原子交换新旧head load() // 原子读取 store() // 原子写入 memory_order_acquire // 获取其他线程发布的数据 memory_order_release // 发布当前线程已经完成的数据另外还要特别注意MPSC 的前提是 Multiple Producer Single Consumer。多个线程可以同时执行Enqueue();但是这个版本不能随意让多个消费者同时执行Dequeue();否则原本“只有一个消费者操作 tail”的设计前提就被破坏了。从前面三篇串起来看线程安全队列的学习路线其实非常清楚第一步 deque mutex ↓ 理解最基本的线程安全 第二步 queue mutex condition_variable ↓ 理解生产者消费者、等待和唤醒 第三步 链表 atomic MPSC ↓ 理解不依赖mutex的并发队列其中mutex更容易编写和理解适合绝大多数常规业务场景而atomic 无锁数据结构实现难度更高需要进一步理解 CAS、内存序以及线程之间的数据可见性一般用于对并发性能要求较高的场景。因此实际开发中并不是看到无锁队列就一定要替换mutex而是应该根据并发量、实现复杂度和实际性能需求进行选择。0voice · GitHub
一、什么是 MPSC 队列前面实现线程安全队列时主要使用std::mutex保护共享队列。例如生产者添加数据std::lock_guardstd::mutex lock(_queueLock); _queue.push(value);消费者获取数据std::lock_guardstd::mutex lock(_queueLock); value _queue.front(); _queue.pop();这种方式很好理解但是每次操作共享队列都需要加锁 ↓ 访问队列 ↓ 解锁当多个线程频繁访问队列时就会出现多个线程竞争同一把锁的情况。于是还可以考虑另一种实现思路不使用传统的mutex保护整个队列而是利用std::atomic提供的原子操作完成线程之间的同步。这就会涉及无锁队列Lock-Free Queue。本篇先介绍其中非常典型的一种MPSC QueueMPSC 全称Multi Producer Single Consumer即Multi Producer 多个生产者 Single Consumer 单个消费者结构可以理解为生产者1 ───┐ │ 生产者2 ───┼──→ MPSC Queue ───→ 消费者 │ 生产者3 ───┘也就是说多个线程可以同时向队列写数据 只有一个线程负责从队列取数据例如日志系统中网络线程 ─────┐ │ 数据库线程 ───┼──→ 日志队列 ───→ 日志写入线程 │ 工作线程 ─────┘多个线程都会产生日志但是可以只安排一个线程真正将日志写入文件。这种场景就非常适合 MPSC 模型。和前面的生产者消费者队列相比mutex队列 生产者 ↓ mutex ↓ queue ↓ mutex ↓ 消费者MPSC 无锁队列则更倾向于生产者 ↓ atomic ↓ 链表节点 ↓ atomic ↓ 消费者这里的“无锁”主要指队列核心同步过程不依赖std::mutex而是使用std::atomic完成线程之间的同步。二、MPSC队列的核心结构Node、head和tail先来看队列内部最关键的数据结构templatetypename T class MPSCQueue { private: struct Node { T* Data; // 当前节点保存的数据 std::atomicNode* Next; // 指向下一个节点 Node() : Data(nullptr), Next(nullptr) {} explicit Node(T* data) : Data(data), Next(nullptr) {} }; std::atomicNode* _head; // 生产者操作的位置 std::atomicNode* _tail; // 消费者操作的位置 };和std::queue不同这里的队列底层实际上是一条单向链表Node ┌─────────────┐ │ Data │ │ Next ───────────→ └─────────────┘每一个节点包含两个东西T* Data;负责保存真正的数据。以及std::atomicNode* Next;保存下一个节点的地址。为什么Next不是普通的Node* Next;而是std::atomicNode* Next;因为生产者和消费者可能同时访问节点之间的连接关系。如果使用普通指针Node* Next;多个线程之间直接读写可能产生数据竞争。因此这里将指针本身声明为原子变量std::atomicNode* Next;整个链表大概是┌──────┐ ┌──────┐ ┌──────┐ │ Node │ ───→ │ Node │ ───→ │ Node │ ───→ nullptr │ 100 │ │ 200 │ │ 300 │ └──────┘ └──────┘ └──────┘除了节点以外还有两个重要指针std::atomicNode* _head; std::atomicNode* _tail;可以先简单理解成_head生产者主要操作的位置 _tail消费者主要操作的位置但这个实现初始化时不会直接让队列完全为空而是先创建一个哑节点Dummy NodeMPSCQueue() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed)) { Node* front _head.load(std::memory_order_relaxed); front-Next.store(nullptr, std::memory_order_relaxed); }刚创建队列时head ↓ ┌───────────┐ tail ─→ │ DummyNode │ ─→ nullptr └───────────┘也就是_head ↓ Dummy ↑ _tail这个 Dummy 节点并不保存真正需要处理的数据。它主要用于统一入队和出队操作减少对第一个元素 最后一个元素 空队列这些特殊情况的额外判断。例如真正加入数据以后Dummy → Node100 → Node200 → Node300Dummy 节点始终可以作为队列结构中的一个过渡节点。这种Dummy Node 哑节点 / 虚拟头节点在链表、队列以及很多无锁数据结构中都非常常见。三、Enqueue多个生产者如何同时添加数据MPSC 最关键的地方就是多个生产者如何在不使用 mutex 的情况下安全入队入队函数如下void Enqueue(T* input) { Node* node new Node(input); // 创建新节点 Node* prevHead _head.exchange(node, std::memory_order_acq_rel); // 原子交换head prevHead-Next.store(node, std::memory_order_release); // 将旧head连接到新节点 }代码只有三句但是这三句就是整个 MPSC 入队操作的核心。首先Node* node new Node(input);为新数据创建一个节点。例如queue.Enqueue(new Count(100));首先创建┌─────────────┐ │ Data 100 │ │ Nextnull │ └─────────────┘接下来Node* prevHead _head.exchange(node, std::memory_order_acq_rel);这里的exchange()是一个非常重要的原子操作。假设_head → Node100现在执行_head.exchange(Node200);它会完成两件事情第一步 把 _head 修改为 Node200 第二步 返回原来的 Node100所以Node* prevHead _head.exchange(node);执行之后prevHead → Node100 _head → Node200关键就在于exchange()整个过程是原子的。假设有两个生产者Producer1 Producer2同时执行_head.exchange(node);不会出现两个线程把_head修改乱掉的情况。原子变量会保证这些操作表现出一个确定的先后顺序。例如最开始_head → Dummy生产者1加入Node100执行Node* prevHead _head.exchange(Node100);得到prevHead → Dummy _head → Node100然后执行prevHead-Next.store(Node100);形成Dummy → Node100接下来生产者2添加Node200执行Node* prevHead _head.exchange(Node200);得到prevHead → Node100 _head → Node200然后prevHead-Next.store(Node200);最终Dummy → Node100 → Node200 ↑ _head如果继续添加 Node300Dummy → Node100 → Node200 → Node300 ↑ _head因此生产者每次实际上做的是创建新节点 ↓ 原子交换 _head ↓ 获得旧 head ↓ 旧 head 的 Next 指向新节点也就是Node* prevHead _head.exchange(node, std::memory_order_acq_rel); prevHead-Next.store(node, std::memory_order_release);这里为什么不用_head node;因为多个生产者可能同时修改_head。普通赋值并不能完成读取旧值 写入新值 返回旧值这一整套不可分割的操作。而exchange()可以原子地完成这一过程。这也是无锁代码中非常常见的写法。这里还涉及两个内存序std::memory_order_acq_rel std::memory_order_release可以先简单理解成exchange(..., acq_rel)既具有 acquire 的读取语义也具有 release 的发布语义。而Next.store(node, std::memory_order_release);表示生产者完成节点内容之后再把这个节点正式发布到链表中。这样消费者后面配合load(std::memory_order_acquire)读取时就能够正确看到生产者已经准备好的节点数据。所以生产者一侧可以概括为准备数据 ↓ release 发布 ↓ 消费者 acquire 获取四、Dequeue单消费者如何从队列中取出数据再来看消费者bool Dequeue(T* result) { Node* tail _tail.load(std::memory_order_relaxed); Node* next tail-Next.load(std::memory_order_acquire); if (!next) return false; result next-Data; _tail.store(next, std::memory_order_release); delete tail; return true; }由于 MPSC 的特点是多个 Producer 一个 Consumer所以真正执行Dequeue()的只有一个消费者线程。首先Node* tail _tail.load(std::memory_order_relaxed);获取当前_tail。假设现在_tail ↓ Dummy → Node100 → Node200那么tail指向Dummy然后Node* next tail-Next.load(std::memory_order_acquire);获取下一个节点。此时tail ↓ Dummy → Node100 ↑ next如果next nullptr说明当前没有可以取出的节点if (!next) return false;如果存在节点result next-Data;把数据返回给调用者。例如next-Data 100那么result → 100接着_tail.store(next, std::memory_order_release);让_tail向后移动。原来_tail ↓ Dummy → Node100 → Node200现在_tail ↓ Dummy → Node100 → Node200然后delete tail;删除原来的旧节点删除 Dummy最终_tail ↓ Node100 → Node200注意这里有一个比较容易第一次看懵的地方。执行result next-Data;取到的是next节点的数据但是delete tail;删除的却是上一个节点为什么不是直接delete next;这是因为当前的next还需要继续充当新的尾部节点。例如Dummy → Node100 → Node200第一次出队取出 Node100 的 Data 删除 Dummy _tail Node100变成Node100 → Node200 ↑ tail第二次出队取出 Node200 的 Data 删除 Node100 _tail Node200所以每次都是读取 next 的数据 删除旧 tail 让 next 成为新的 tail这其实也是 Dummy Node 设计带来的好处。整个出队流程可以总结成读取 tail ↓ 找到 tail-Next ↓ Next不存在 ↓ 是 队列暂时没有数据 ↓ 否 获取 next-Data ↓ tail移动到next ↓ 删除旧tail ↓ 返回true这里生产者发布节点时prevHead-Next.store(node, std::memory_order_release);消费者读取节点时tail-Next.load(std::memory_order_acquire);正好构成Producer release ↓ 节点数据发布 ↓ acquire Consumer这样消费者看到Next ! nullptr以后也能够正确看到这个节点之前已经准备好的Data这就是 acquire/release 在这个队列中的一个实际应用。五、完整MPSC队列与多线程测试把前面的结构组合起来可以得到一个简化后的 MPSC 非侵入式队列#ifndef MPSC_QUEUE_H #define MPSC_QUEUE_H #include atomic templatetypename T class MPSCQueue { private: struct Node { T* Data; std::atomicNode* Next; Node() : Data(nullptr), Next(nullptr) {} explicit Node(T* data) : Data(data), Next(nullptr) {} }; std::atomicNode* _head; std::atomicNode* _tail; public: // 创建Dummy节点head和tail最开始都指向它 MPSCQueue() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed)) {} // 禁止拷贝 MPSCQueue(const MPSCQueue) delete; MPSCQueue operator(const MPSCQueue) delete; ~MPSCQueue() { T* data; // 清理队列中还没有取出的数据 while (Dequeue(data)) delete data; // 删除最后留下的Dummy节点 Node* tail _tail.load(std::memory_order_relaxed); delete tail; } // 多个生产者可以调用Enqueue void Enqueue(T* input) { Node* node new Node(input); // 原子替换head并获得之前的head Node* prevHead _head.exchange(node, std::memory_order_acq_rel); // 将旧head连接到新节点 prevHead-Next.store(node, std::memory_order_release); } // 只允许一个消费者调用Dequeue bool Dequeue(T* result) { Node* tail _tail.load(std::memory_order_relaxed); // 获取tail后面的节点 Node* next tail-Next.load(std::memory_order_acquire); // 没有数据 if (!next) return false; // 获取数据 result next-Data; // tail向后移动 _tail.store(next, std::memory_order_release); // 删除旧tail delete tail; return true; } }; #endif测试时先定义一个简单的数据类型struct Count { explicit Count(int value) : v(value) {} int v; };然后创建MPSCQueueCount queue;两个生产者不断向队列添加数据一个消费者负责获取#include atomic #include iostream #include thread int main() { MPSCQueueCount queue; std::atomicint producerFinished{0}; // 生产者1 std::thread producer1([]() { queue.Enqueue(new Count(100)); queue.Enqueue(new Count(200)); queue.Enqueue(new Count(300)); queue.Enqueue(new Count(400)); producerFinished; }); // 生产者2 std::thread producer2([]() { queue.Enqueue(new Count(500)); queue.Enqueue(new Count(600)); queue.Enqueue(new Count(700)); queue.Enqueue(new Count(800)); producerFinished; }); // MPSC只有一个消费者 std::thread consumer([]() { Count* data nullptr; while (true) { if (queue.Dequeue(data)) { std::cout pop : data-v std::endl; delete data; continue; } // 两个生产者都结束并且当前已经取不到数据则结束消费 if (producerFinished.load() 2) break; std::this_thread::yield(); } }); producer1.join(); producer2.join(); consumer.join(); return 0; }这里线程关系非常清楚producer1 │ ├──100 ├──200 ├──300 └──400 │ ↓ ┌─────────┐ │ │ │ MPSC │ ─────→ consumer │ Queue │ │ │ └─────────┘ ↑ │ ┌──500 ├──600 ├──700 └──800 producer2和上一篇最大的区别是上一篇使用std::mutex std::condition_variable实现的是加锁 等待 唤醒这一篇则主要依靠std::atomicNode*以及exchange() load() store()维护链表。其中最关键的入队操作只有Node* prevHead _head.exchange(node, std::memory_order_acq_rel); prevHead-Next.store(node, std::memory_order_release);而消费者最关键的逻辑是Node* next tail-Next.load(std::memory_order_acquire);因此整个 MPSC 队列可以概括成多个生产者 ↓ atomic exchange ↓ 不断向链表尾部连接新节点 ↓ MPSC Queue ↓ 单个消费者 ↓ 沿着 Next 顺序取出数据学习这一部分最需要掌握的并不是一开始就把整个无锁队列背下来而是理解下面几个知识点之间的关系std::atomicNode* // 原子指针 exchange() // 原子交换新旧head load() // 原子读取 store() // 原子写入 memory_order_acquire // 获取其他线程发布的数据 memory_order_release // 发布当前线程已经完成的数据另外还要特别注意MPSC 的前提是 Multiple Producer Single Consumer。多个线程可以同时执行Enqueue();但是这个版本不能随意让多个消费者同时执行Dequeue();否则原本“只有一个消费者操作 tail”的设计前提就被破坏了。从前面三篇串起来看线程安全队列的学习路线其实非常清楚第一步 deque mutex ↓ 理解最基本的线程安全 第二步 queue mutex condition_variable ↓ 理解生产者消费者、等待和唤醒 第三步 链表 atomic MPSC ↓ 理解不依赖mutex的并发队列其中mutex更容易编写和理解适合绝大多数常规业务场景而atomic 无锁数据结构实现难度更高需要进一步理解 CAS、内存序以及线程之间的数据可见性一般用于对并发性能要求较高的场景。因此实际开发中并不是看到无锁队列就一定要替换mutex而是应该根据并发量、实现复杂度和实际性能需求进行选择。0voice · GitHub
