1.环形队列是什么

队列是一种常用的数据结构，这种结构保证了数据是按照“先进先出”的原则进行操作的，即最先进去的元素也是最先出来的元素.环形队列是一种特殊的队列结构，保证了元素也是先进先出的，但与一般队列的区别是，他们是环形的，即队列头部的上个元素是队列尾部，通常是容纳元素数固定的一个闭环。

C代码实现见:https://github.com/dodng/fast_ring_queue

2.环形队列的优点

　　1.保证元素是先进先出的

是由队列的性质保证的，在环形队列中通过对队列的顺序访问保证。

　　2.元素空间可以重复利用

因为一般的环形队列都是一个元素数固定的一个闭环，可以在环形队列初始化的时候分配好确定的内存空间，当进队或出队时只需要返回指定元素内存空间的地址即可，这些内存空间可以重复利用，避免频繁内存分配和释放的开销。

　　3.为多线程数据通信提供了一种高效的机制。

在最典型的生产者消费者模型中，如果引入环形队列，那么生成者只需要生成“东西”然后放到环形队列中即可，而消费者只需要从环形队列里取“东西”并且消费即可，没有任何锁或者等待，巧妙的高效实现了多线程数据通信。

3.环形队列的工作场景

一般应用于需要高效且频繁进行多线程通信传递数据的场景，例如：linux捕包、发包等等，（linux系统中对PACKET_RX_RING和PACKET_TX_RING的支持实质就是内核实现的一种环形队列）

实际环形队列在工作时有3种情况：

　　3.1 入队速度=出队速度

这是环形队列的常态，即入队速度和出队速度大致一样，即使某个突然时刻入队速度陡然变高或者出队速度陡然变低，都能通过队列这个缓冲区把这些数据先存起来，等到能处理的时候再处理。

　　3.2 入队速度>出队速度

在这种情况下，队列“写入”的速度>“读取”的速度，想象当这种状态持续一段时间之后，队列中大多数全是写入但没读取的元素，当又一个新的元素产生时，可以把这个新元素drop掉或者放在另一个缓冲区保存起来，这种情况的出现不是个好事情，说明你需要对出队处理元素的算法或逻辑优化处理速度了。

　　3.3 入队速度<出队速度

在这种情况下，队列“读取”速度>“写入”速度，这种情况说明程序出队处理元素的速度很快，这是比较好的情况，唯一不足的是读取队列的时候可能经常会轮询队列是否有新的元素，造成cpu占用过高。

4.无锁环形队列的实现

　　4.1环形队列的存储结构

链表和线性表都是可以的，但几乎都用线性表实现，比链表快很多，原因也是显而易见的，因为访问链表需要挨个遍历。

　　4.2读写index

有2个index很重要，一个是写入index，标示了当前可以写入元素的index，入队时使用。一个是读取index，标示了当前可以读取元素的index，出队时使用。

4.3元素状态切换

有种很巧妙的方法，就是在队列中每个元素的头部加一个元素标示字段，标示这个元素是可读还是可写，而这个的关键就在于何时设置元素的可读可写状态，参照linux内核实现原理，当这个元素读取完之后，要设置可写状态，当这个元素写入完成之后，要设置可读状态。

5.实际测试效果

在CentOS 5.5 (cpu每核主频1200MHz)设备上简单测试了一下。环形队列大小为10000，元素数据大小为4字节，每秒可以写入并读取的元素是250万左右，即250pps.

C代码实现见:https://github.com/dodng/fast_ring_queue

======================================================================================================

第二篇：彻底理解高性能无锁队列

问题引入：

一个生产者，多个消费者的队列，如果是你，你回怎么设计？

想必拿到这个问题，更多的人脑海中已经浮现了一把锁；我也是的，那我们就从浅入深的来看看高性能的无锁队列是怎么一步一步的演化开来的。

1、低效的实现队列

编写多线程的时候，往往会发生资源竞争的现象，导致我们不得不加锁去保护变量，但在这个的同时也对性能造成了一定的损耗。

如图所示：

这种设计方式的话，我们是不可避免加锁操作的，因为其本身就不是线程安全的。

流程：

生产者放入队列中时，加锁，数据输入后完成加锁操作；然后剩余线程进行争锁操作，进行取队列数据操作；

简单实现：

template<class T>
class SimpleQueue
{
public:SimpleQueue() {}~SimpleQueue(){}void Push(T val){_mutex.lock();_q.push(move(val));_mutex.unlock();}T Get(){_mutex.lock();if (_q.empty()){_mutex.unlock();return 0;}T val = _q.front();_q.pop();_mutex.unlock();return val;}private:mutex _mutex;queue<T>_q;
};

这个就是比较简单，同事性能较差的一种方案；

既然我们提到了高性能，那么这种操作是不不符合我们需求的，那还有什么更好的方案提供跟高的性能吗？

很显然的一个操作：去锁化-----也就是常说的无锁队列

2、无锁队列

其实有锁和无锁就是我们平时所说的乐观锁和悲观锁：

加锁是一种悲观的策略，它总是认为每次访问共享资源的时候，总会发生冲突，所以宁愿牺牲性能（时间）来保证数据安全。
无锁是一种乐观的策略，它假设线程访问共享资源不会发生冲突，所以不需要加锁，因此线程将不断执行，不需要停止。一旦碰到冲突，就重试当前操作直到没有冲突为止。

无锁的策略使用一种叫做比较交换的技术（CAS Compare And Swap）来鉴别线程冲突，一旦检测到冲突产生，就重试当前操作直到没有冲突为止。

CAS是系统原语，CAS操作是一条CPU的原子指令，所以不会有线程安全问题。

CAS 的伪码:

template <class T>
bool CAS(T* addr, T expected, T value) 
{if (*addr == expected) {*addr = value;return true;}return false;
}

CAS 将 expected 与一个内存地址进行比较，如果比较成功，就将内存内容替换为 new 。当前大多数机器都在硬件级实现了这个操作，在 Inter 处理器上这个操作是 CMPXCHG ，因而 CAS 是一个最基础的原子操作。

GCC4.1+版本中支持CAS的原子操作,API接口如下：

**bool** __sync_bool_compare_and_swap(type *ptr, type oldval type newval, ...)type __sync_val_compare_and_swap(type *ptr, type oldval type newval, ...)#include <algorithm>
#include <vector>
#include <string>
#include <unordered_map>
#include <queue>
#include <functional>
#include <stack>
#include <iostream>
#include <unistd.h>
#include <thread>
#include <list>using namespace std;/*
*   说明：基于CAS封装的无锁List。
*/
template <typename T>
class LockfreeList
{
private:std::list<T> list;private:volatile int mutex;int lock;int unlock;public:LockfreeList() :mutex(0), lock(0), unlock(1) {};~LockfreeList() {};void Lock(){while (!__sync_bool_compare_and_swap(&mutex, lock, 1)){usleep(1);}}void Unlock(){__sync_bool_compare_and_swap(&mutex, unlock, 0);}void Push(T data){Lock();list.push_back(data);Unlock();}T Front(){Lock();T data = list.front();Unlock();return data;}void PopFront(){Lock();list.pop_front();Unlock();}bool IsEmpty(){Lock();if (list.empty()){Unlock();return true;}else{Unlock();return false;}}bool Find(T data){typename std::list<T>::iterator it;Lock();for (it = list.begin(); it != list.end(); ++it){if (*it == data){Unlock();return true;}}Unlock();return false;}
};LockfreeList<int> LF;thread_local int num = 1;
//生产者
void Producer()
{while(true){num++;cout<<"num push:"<<num<<endl;LF.Push(num);sleep(2);}
}//消费者
void Customer()
{while(true){if (!LF.IsEmpty()){cout <<"num get " <<LF.Front() <<endl;LF.PopFront();}sleep(1);}
}int main()
{thread t1(Producer);thread t2(Customer);thread t3(Customer);t1.join();t2.join();t3.join();return 0;
}

在C++11 中出现了CAS的用法，也为我们提供了API；

/*
* @brief：compare & swap(CAS)。如果等于expect则swap，否则就返回--是否交换成功, 注意expect如果不相等，会把当前值写入到expected里面。
* 相比于strong，weak可能会出现[spurious wakeup](<http://en.wikipedia.org/wiki/Spurious_wakeup>).
* @param          若x等于expect,则设置为desired 返回true，
*                 否则最新值写入expect，返回false
*/
class atomic {bool compare_exchange_strong(T& expect /*用来比较的值*/, T desired/*用来设置的值*/)bool compare_exchange_weak(T& expect, T desired)
}

其实在CAS中，还有一种异常产生，也就是常说的ABA的现象。所谓ABA现象就是当前现象期望值是A,某个线程将A改为B，另外线程将B改为A，导致当前线程误以为还是原来的值，然后操作就会导致一些异常出现。

这里我们可以借用数据库乐观锁的方式，维护一个全局的版本号或者是标志，每次修改的时候需要期望值和内存值相等并且标识也没有发生改变的时候采取更新值。

无锁（CAS）本身编程就不是很友好，如果没有彻底掌握，最好还是使用锁去编写。CAS 更多的是一种思想，也是实现高性能编程的一种途径，目前已经有一些开源级别的无锁库可以提供我们使用，也许这些才是我们最好的选择。

=====================================================================================