[数据结构]

2023-07-05 08:21| 来源: 网络整理| 查看: 265

无锁队列

写这篇博客前想声名以下几点。第一，这篇文章重点内容是关于无锁队列如何实现，并不会深入讲解底层的CAS机制。原因就是第二条，不知道在看博客的你是否在搜索框中输入过"无锁队列"关键字，你点开居然会惊讶的发现每一篇居然都是那么的相似，一直不理解写博客A抄B，B抄C只是为了骗访客吗？

这篇被抄来抄去的博客就是酷壳陈皓老师的原创，老师的博客中深入讲解了无锁队列的原理。只是缺少最后一份代码的实现，既然有大牛级别的讲解，那我们就完成那部分没有实现的部分。

还不曾知道无锁队列底层CAS机制的同学可以看看陈老师和程序员小灰的这两篇文章：无锁队列的实现和什么是CAS机制

无锁队列基本原理

其实CAS机制并不难理解，你只需要理解CAS机制中的三个关键词。分别为原本值、期望值、更新值(名词)。这三个值配合起来用一句话概括为：如果原本值和期望值相同，那么就将原本值修改为更新值

在这里插入图片描述从图中可以看出，CAS机制并不难理解，那么下面就一起看看无锁队列的思想吧。

无锁队列具体实现

我们这里需要使用C++11中引入的atomic类，这个类被意为原子的操作，函数接口很容易使用，我们先来看无锁队列实现思想，然后结合代码来理解会更容易

无锁队列基本结构

对每一个指针操作时都需要是原子的，也就是指针的指向要么发生变化，要么没有变化，不会产生第三种结果。每个队列需要有一个哑节点，这样能更好的判断队列是否为空。

#include using namespace std; template class LockFreeQueue { struct Node { T _value; atomic _next; Node(const T& v) :_value(v) ,_next(nullptr) {} }; public: LockFreeQueue() { _head = _tail = new Node(T()); } ~LockFreeQueue() { Node* cur = _head; while(cur) { Node* next = cur->_next; delete cur; cur = next; } } private: atomic _head; atomic _tail; };

无锁队列如何入队列？

在这里插入图片描述下面介绍一下atomic类型的两个函数操作

//取出当前atomic对象指向的值 T load (memory_order sync = memory_order_seq_cst) const volatile noexcept; //如果内存中的值和参数一相同，那么替换为参数二，成功返回true bool compare_exchange_weak (T& expected, T val, memory_order sync = memory_order_seq_cst) volatile noexcept;

两个函数看起来很复杂，用起来其实超级简单，现在一起来看对无锁队列的插入代码块：

void Enqueue(const T& x) { Node* newnode = new Node(x);//要插入的新值 Node* oldtail = nullptr;//旧的尾节点 Node* nullnode = nullptr;//空节点 do { oldtail = _tail.load();//用load取出当前_tail节点的值 } //如果当前tail节点的下一个节点是空，也就是等于参数一，那么修改为参数二 while (oldtail->_next.compare_exchange_weak(nullnode, newnode) != true); //由于现在的真正的尾节点是newnode，所以将_tail节点更新为newnode _tail.compare_exchange_weak(oldtail, newnode); }

这里要明白，while条件里的CAS语句是为了插入这个新节点，而最后一句是为了将控制整个队列的_tail指针修改为指向新尾部

无锁队列如何出队列？

在这里插入图片描述出队列比插入更简单，那么废话不多说，直接看看出队列如何实现：

T DeQueue() { Node* oldhead = _head.load();//取出当前的头节点，也就是哑节点 T ret; do { Node* next = oldhead->_next;//取出头节点的下一个节点，此节点中有我们想要的值 if(next == nullptr) { return T(); } else { ret = next->_value;//取走值 } } //将头节点(哑节点)修改为下一个节点 while(_head.compare_exchange_weak(oldhead,oldhead->_next) != true); delete oldhead; return ret; }

判断队列是否为空和其他问题

判断是否为空，直接判断头部和尾部是否指向同一个节点

bool Empty() { return _head.load() == _tail.load(); }

因为我们的无锁队列是不能够被拷贝的，所以我们使用delete关键字直接将拷贝构造函数删除掉

LockFreeQueue(const LockFreeQueue&) = delete;

到这里，我们的无锁队列就实现完成了，之所以这么简单，还是因为C++库为我们提供了atomic这个类。无锁队列这里重在理解他的如何使用CAS的思想去更新节点的。

无锁队列全部源码和测试

经过笔者多次测试，发现无锁队列总体来说还是不错的，比加锁的队列能快一倍左右(第一行为无锁队列)，这是插入10w数据的一次的结果，基本在这两个数字中波动。但是无锁队列对于线程数量的选择很关键。随着线程数增大，最后会比加锁队列更差，原因不难理解，CAS机制本身还是太消耗CPU资源了。

在这里插入图片描述

#pragma once #include #include #include #include #include #include using namespace std; template class LockFreeQueue { struct Node { T _value; std::atomic _next; Node(const T& x) : _value(x) , _next(nullptr) {} }; public: LockFreeQueue() { _head = _tail = new Node(T()); } LockFreeQueue(const LockFreeQueue&) = delete; ~LockFreeQueue() { Node* cur = _head; while (cur) { Node* next = cur->_next; delete cur; cur = next; } } void Enqueue(const T& x) { Node* newnode = new Node(x); Node* oldtail = nullptr; Node* nullnode = nullptr; do { oldtail = _tail.load(); } while (oldtail->_next.compare_exchange_weak(nullnode, newnode) != true); _tail.compare_exchange_weak(oldtail, newnode); } T Dequeue() { Node* oldhead = _head.load(); T headvalue; do { Node* next = oldhead->_next; if (next == nullptr) { return T(); } else { headvalue = next->_value; } } while (_head.compare_exchange_weak(oldhead, oldhead->_next) != true); delete oldhead; return headvalue; } bool Empty() { return _head.load() == _tail.load(); } private: std::atomic _head; std::atomic _tail; }; mutex mutx; int main() { LockFreeQueue lq; queue nq; atomic n = 100000; //测试无锁队列 { size_t costtime = 0; vector threads; for (size_t i = 0; i size_t begin = 0, end = 0; begin = clock(); for (size_t j = 0; j e.join(); } cout threads.push_back(thread([&]() { size_t begin = 0, end = 0; begin = clock(); for (size_t j = 0; j e.join(); } cout

【本文地址】

[数据结构]

[数据结构]

今日新闻

推荐新闻