无锁队列（Lock-Free Queue）

无锁队列原理无锁队列Lock-Free Queue是一种基于无锁编程Lock-Free Programming技术实现的并发数据结构。它的核心思想是1.基础原理使用 CASCompare-And-Swap比较并交换等原子操作来保证线程安全内存序 (Memory Ordering): 确保多线程间的内存可见性;避免使用互斥锁mutex消除了线程阻塞和上下文切换的开销通过原子变量的忙等待spin来实现并发控制2.CAS 操作CAS 是无锁编程的基础原语包含三个操作数内存值 V预期值 A新值 B当且仅当 V A 时将 V 更新为 B否则不更新。整个操作是原子的。3.ABA 问题CAS 存在 ABA 问题如果内存值从 A→B→ACAS 会误认为值未被修改。解决方案使用带版本号的 CAS如 std::atomicstd::pairT, int使用双指针技术如 Michael-Scott 队列4.典型实现 - Michael Scott 算法最常见的无锁队列是 Michael Scott 算法其核心结构Head 指针 → [Node] → [Node] → [Node] → Tail 指针入队操作 (Enqueue):1. 创建新节点2. 循环直到成功- 获取当前 Tail 节点- 将新节点设为 Tail 的 next- 使用 CAS 尝试更新 Tail 指针到新节点出队操作 (Dequeue):1. 循环直到成功- 获取当前 Head 节点- 获取 Head 的 next 节点- 使用 CAS 尝试移动 Head 指针到 next- 返回原 Head 的数据5典型实现-基于循环数组的实现使用原子索引和循环数组.enqueue() {current tail.load(acquire)next (current 1) % capacityif CAS(tail, current, next) {array[current] valuereturn}}C实现创建文件lockfree_queue.h 和 lockfree_queue_test.cpp具体实现见附件。实现原理1.核心技术CAS (Compare-And-Swap) - 原子操作CAS(预期值, 新值) {if (内存值 预期值) {内存值 新值;return 成功;} else {return 失败;}}2.Michael-Scott 算法关键点┌───────────────┬──────────────────────────────────────┐│ 特性 │ 说明 │├───────────────┼──────────────────────────────────────┤│ Sentinel 节点 │ 头节点使用哨兵节点简化边界条件处理 │├───────────────┼──────────────────────────────────────┤│ 双指针 │ head 和 tail 指针分别指向队列的两端 │├───────────────┼──────────────────────────────────────┤│ 两阶段操作 │ 先通过 CAS 链接节点再移动指针 │├───────────────┼──────────────────────────────────────┤│ 协助机制 │ 落后的线程帮助推进指针保证进度 │└───────────────┴──────────────────────────────────────┘3.入队流程 (push)3.1. 读取 tail 和 tail-next3.2. 如果 tail-next nullptr说明是尾部3.3. CAS 将新节点链接到 tail-next3.4. CAS 尝试移动 tail 到新节点4.出队流程 (pop)4.1.读取 head、tail 和 head-next4.2.如果 head tail 且 next nullptr队列为空4.3.读取 next 节点的值4.4.CAS 将 head 移动到 next4.5.释放旧的 head 节点5.内存序说明┌──────────────────────┬──────────────────────────────────┐│ 内存序 │ 作用 │├──────────────────────┼──────────────────────────────────┤│ memory_order_acquire │ 读操作防止后续操作重排序到前面 │├──────────────────────┼──────────────────────────────────┤│ memory_order_release │ 写操作防止前面操作重排序到后面 │├──────────────────────┼──────────────────────────────────┤│ memory_order_relaxed │ 无同步仅原子性 │└──────────────────────┴──────────────────────────────────┘主要接口push(T value) - 入队pop() - 阻塞出队忙等try_pop() - 非阻塞出队empty() - 是否为空size() - 队列大小编译运行需要 C17g -stdc17 -pthread lockfree_queue_test.cpp -o lockfree_queue_test./lockfree_queue_test附件/** * file lockfree_queue.h * brief Michael-Lock-Free Queue Implementation * * 基于 CAS 的无锁队列支持多生产者多消费者MPMC */ #ifndef LOCKFREE_QUEUE_H #define LOCKFREE_QUEUE_H #include atomic #include memory #include utility template typename T class LockFreeQueue { public: LockFreeQueue() : head_(new Node()), tail_(head_.load()) { head_.load()-next_.store(nullptr, std::memory_order_relaxed); } ~LockFreeQueue() { while (pop_impl()) {} delete head_.load(); } // 禁用拷贝 LockFreeQueue(const LockFreeQueue) delete; LockFreeQueue operator(const LockFreeQueue) delete; // 禁用移动 LockFreeQueue(LockFreeQueue) delete; LockFreeQueue operator(LockFreeQueue) delete; /** * brief 入队操作线程安全 * param value 要入队的元素 */ void push(T value) { Node* new_node new Node(std::move(value)); while (true) { Node* tail tail_.load(std::memory_order_acquire); Node* next tail-next_.load(std::memory_order_acquire); // 检查 tail 是否仍然是尾节点 if (tail tail_.load(std::memory_order_acquire)) { if (next nullptr) { // 尝试将新节点链接到尾节点 if (tail-next_.compare_exchange_weak(next, new_node, std::memory_order_release, std::memory_order_acquire)) { // 成功链接尝试移动 tail // 使用 compare_exchange_strong 避免 ABA 问题 tail_.compare_exchange_strong(tail, new_node, std::memory_order_release, std::memory_order_acquire); return; } } else { // tail 已经落后帮助推进 tail tail_.compare_exchange_weak(tail, next, std::memory_order_release, std::memory_order_acquire); } } } } /** * brief 出队操作线程安全 * return 包含元素的 optional如果队列为空则返回空 */ std::optionalT pop() { while (true) { Node* head head_.load(std::memory_order_acquire); Node* tail tail_.load(std::memory_order_acquire); Node* next head-next_.load(std::memory_order_acquire); // 检查 head 是否仍然是头节点 if (head head_.load(std::memory_order_acquire)) { if (head tail) { // 可能是空队列或 tail 落后 if (next nullptr) { // 队列为空 return std::nullopt; } // tail 落后帮助推进 tail_.compare_exchange_weak(tail, next, std::memory_order_release, std::memory_order_acquire); } else { // 读取下一个节点的值 T value std::move(next-value_); // 尝试移动 head 指针 if (head_.compare_exchange_weak(head, next, std::memory_order_release, std::memory_order_acquire)) { // 成功删除旧的 sentinel 节点 delete head; return value; } // 失败重试 } } } } /** * brief 尝试出队非阻塞 * return 包含元素的 optional */ std::optionalT try_pop() { Node* head head_.load(std::memory_order_acquire); Node* next head-next_.load(std::memory_order_acquire); if (next nullptr) { return std::nullopt; } T value std::move(next-value_); if (head_.compare_exchange_strong(head, next, std::memory_order_release, std::memory_order_acquire)) { delete head; return value; } return std::nullopt; } /** * brief 检查队列是否为空 * note 返回结果可能立即过期 */ bool empty() const { Node* head head_.load(std::memory_order_acquire); Node* next head-next_.load(std::memory_order_acquire); return next nullptr; } /** * brief 获取队列大小近似值 * note 返回结果可能立即过期 */ size_t size() const { size_t count 0; Node* head head_.load(std::memory_order_acquire); Node* next head-next_.load(std::memory_order_acquire); while (next ! nullptr) { count; head next; next head-next_.load(std::memory_order_acquire); } return count; } private: struct Node { std::atomicNode* next_; T value_; Node() : next_(nullptr) {} template typename U explicit Node(U value) : next_(nullptr), value_(std::forwardU(value)) {} }; std::atomicNode* head_; std::atomicNode* tail_; }; #endif // LOCKFREE_QUEUE_H/** * file lockfree_queue_test.cpp * brief LockFreeQueue 测试程序 */ #include lockfree_queue.h #include iostream #include thread #include vector #include chrono #include atomic #include functional // 测试基本入队出队 void test_basic() { std::cout 基本功能测试 std::endl; LockFreeQueueint queue; // 入队 for (int i 1; i 10; i) { queue.push(i); } std::cout 入队 10 个元素后队列大小: queue.size() std::endl; // 出队 std::cout 出队顺序: ; while (auto val queue.pop()) { std::cout *val ; } std::cout std::endl; std::cout 出队后队列大小: queue.size() std::endl; std::cout 队列是否为空: (queue.empty() ? 是 : 否) std::endl; std::cout std::endl; } // 测试多生产者多消费者 void test_mpmc() { std::cout 多生产者多消费者测试 std::endl; LockFreeQueueint queue; const int num_producers 4; const int num_consumers 4; const int items_per_producer 1000; const int total_items num_producers * items_per_producer; std::atomicint produced_count{0}; std::atomicint consumed_count{0}; std::atomicbool production_done{false}; std::vectorint consumed_data; std::mutex data_mutex; // 仅用于保护输出不在队列中使用 // 启动生产者 std::vectorstd::thread producers; for (int p 0; p num_producers; p) { producers.emplace_back([queue, produced_count, items_per_producer, p] { for (int i 0; i items_per_producer; i) { queue.push(p * 10000 i); produced_count.fetch_add(1, std::memory_order_relaxed); } }); } // 启动消费者 std::vectorstd::thread consumers; for (int c 0; c num_consumers; c) { consumers.emplace_back([queue, consumed_count, consumed_data, data_mutex, production_done] { while (true) { auto val queue.try_pop(); if (val.has_value()) { consumed_count.fetch_add(1, std::memory_order_relaxed); std::lock_guardstd::mutex lock(data_mutex); consumed_data.push_back(*val); } else { // 队列为空检查是否所有生产已完成且队列已空 if (production_done.load() queue.empty()) { break; } // 队列为空但生产仍在进行短暂休眠后重试 std::this_thread::yield(); } } }); } // 等待生产者完成 for (auto t : producers) { t.join(); } produced_count.store(total_items); // 确保计数正确 production_done.store(true); // 通知消费者生产已完成 // 等待消费者完成 for (auto t : consumers) { t.join(); } std::cout 生产数量: produced_count.load() std::endl; std::cout 消费数量: consumed_count.load() std::endl; std::cout 数据一致性: (produced_count.load() consumed_count.load() ? 通过 : 失败) std::endl; std::cout std::endl; } // 性能测试 void test_performance() { std::cout 性能测试 std::endl; LockFreeQueueint queue; const int num_items 100000; // 生产者 auto start std::chrono::steady_clock::now(); for (int i 0; i num_items; i) { queue.push(i); } auto produce_end std::chrono::steady_clock::now(); // 消费者 int count 0; while (queue.pop()) { count; } auto consume_end std::chrono::steady_clock::now(); auto produce_time std::chrono::duration_caststd::chrono::microseconds( produce_end - start).count(); auto consume_time std::chrono::duration_caststd::chrono::microseconds( consume_end - produce_end).count(); std::cout 入队 num_items 个元素耗时: (produce_time / 1000.0) ms std::endl; std::cout 出队 num_items 个元素耗时: (consume_time / 1000.0) ms std::endl; std::cout std::endl; } // 测试 try_pop非阻塞 void test_try_pop() { std::cout try_pop 测试 std::endl; LockFreeQueueint queue; // 空队列测试 auto result1 queue.try_pop(); std::cout 空队列 try_pop: (result1.has_value() ? 失败 : 成功(返回空)) std::endl; // 入队后测试 queue.push(42); auto result2 queue.try_pop(); std::cout 有数据 try_pop: (result2.has_value() ? 成功 : 失败) , 值: (result2.has_value() ? std::to_string(*result2) : none) std::endl; // 再次 try_pop应返回空 auto result3 queue.try_pop(); std::cout 再次 try_pop: (result3.has_value() ? 失败 : 成功(返回空)) std::endl; std::cout std::endl; } int main() { test_basic(); test_try_pop(); test_performance(); test_mpmc(); std::cout 所有测试完成 std::endl; return 0; }