1. 什么是无锁数据结构？

锁的本质是阻止其他线程进入锁住的临界区，当一个线程在临界区中休眠，其他线程的操作也会被卡在临界区外（锁的根本意图就是杜绝并发功能，是阻塞型数据结构）。而无锁数据结构要求总有一个线程能够真正推进事情的进展，而不是空转，也就是说即使一些线程在任意位置休眠，其他线程也能完成操作并返回，这也说明任何时候都不存在锁住的临界区。

无锁数据结构不一定更快，因为常常需要很多原子操作，每个原子操作都有额外开销并可能涉及 CPU 和缓存的竞争。

1.无锁数据结构的优点：

最大限度地实现并发：

还是那句话，锁的根本意图就是杜绝并发功能，而无锁数据结构总存在某个线程能执行下一步操作（不存在锁的临界区导致其他线程被堵塞的问题）

代码的健壮性：

假设数据结构的写操作受锁保护，如果某一线程在持锁期间终止，那么该数据结构只完成了部分改动，且此后没办法修补。因为持锁期间，线程会对共享数据结构执行一系列被锁保护的操作，其他线程无法访问数据结构或观察到其部分修改状态，如果线程在操作完成之前终止（例如异常退出），锁会释放，但数据结构可能处于不一致或部分修改的状态，而剩下的部分操作没有其他线程可以接管和恢复操作，因为锁没有记录操作的上下文。 但是在无锁数据结构中，即使某线程操作无锁数据时意外终结，但丢失的数据仅限于它本身持有的部分，其他的数据仍然完好，能被其他线程正常处理（因为原子操作不能被分割，要么成功修改数据，要么失败保持原状态不变，所以即使线程终止，也不会留下半完成的修改）。

2.无锁数据结构的缺点：

难度大：

对无锁数据结构执行写操作的难度高于带锁的数据结构，主要因为无锁数据结构需要在没有锁的情况下依靠复杂的算法和原子操作（如CAS，就是compare_exchange_strong）来保证线程安全。写操作必须确保全局一致性，处理并发冲突，并设计有效的重试机制，同时解决诸如ABA问题等细节。而带锁数据结构只需通过互斥锁避免并发，逻辑相对简单，因此无锁写操作的实现通常更加复杂且易出错。

活锁

由于无锁数据结构完全不含锁，因此不存在死锁问题，但活锁（live lock）反而有可能出现。假设两个线程同时修改同一份数据结构，若他们所做的改动都导致对方从头开始操作，那双方就会反复循环，不断重试，这种现象即为活锁。与死锁不同，活锁中的线程不会被阻塞，它们会持续执行某些操作，但由于逻辑错误或相互之间的干扰，始终无法达到预期的目标。

2. 环形队列

环形队列是多线程无锁并发执行时用到的，一次往队列中写入一个事件，队列只记录事件相关数据的指针，另外使用原子操作来记录读取这个指针，迅速、安全。因为指针占空间小而且一致，所以可以直接使用数组来保存它们。

而环形队列有以下两个好处：

• 成环的队列大小是固定的，可以循环复用

• 通过移动头和尾就能实现数据的插入和取出

一个环形结构示意图如下所示：

• 环形队列是队列的一种数据结构，在队头出队, 队尾入队；

• 只是环形队列的大小是确定的， 不能进行一个长度的增加，当你把一个环形队列创建好之后，它能存放的元素个数是确定的；

• 虽然环形队列在逻辑上是环形的，但在物理上是一个定长的数组；

• 一般我们实现这个环形队列是通过一个连续的结构来实现的;

• 环形队列在逻辑上形成一个环形的变化，主要是当头尾指针当走到连续空间的末尾的时候，它会做一个重置的操作。

• 如上图所示，当队列为空的时候，头指针和尾指针指向同一个区域；

• 当插入一个数据之后，队列size变为1，尾指针Q.rear + 1向前移动到下一个扇区，头指针Q.front存储队列的第一个数据，并始终指向该区域（如果不pop数据的话）；

• 当pop出一个数据后，头指针Q.front + 1 向前移动到下一个扇区，如果 front == rear 表示队列为空。注意：当数据被pop出队列后，仅仅只是头指针变化，而数据其实仍然留在内存原处不用处理，当插入新数据时会将这个内存原本的数据覆盖掉；

• 当尾指针 rear + 1 % 队列长度 == front 时，表示队列为满。

3. 实现线程安全的环形队列

在本节中，我们通过互斥量和原子操作分别实现有锁环形队列和无锁环形队列。

3.1 实现有锁环形队列

代码如下：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

#include #include #include  template<typename T, size_t Cap>class CircularQueLk :private std::allocator{public: CircularQueLk() :_max_size(Cap + 1),_data(std::allocator::allocate(_max_size)), _head(0), _tail(0) {} CircularQueLk(const CircularQueLk&) = delete; CircularQueLk& operator = (const CircularQueLk&) volatile = delete; CircularQueLk& operator = (const CircularQueLk&) = delete;  ~CircularQueLk() { //循环销毁 std::lock_guard<std::mutex>  lock(_mtx); //调用内部元素的析构函数 while (_head != _tail) { std::allocator::destroy(_data + _head); _head = (_head+1)%_max_size; } //调用回收操作 std::allocator::deallocate(_data, _max_size); }  //先实现一个可变参数列表版本的插入函数最为基准函数 template <typename ...Args> bool emplace(Args && ... args) { std::lock_guard<std::mutex> lock(_mtx); //判断队列是否满了 if ((_tail + 1) % _max_size == _head) { std::cout << "circular que full ! " << std::endl; return false; } //在尾部位置构造一个T类型的对象，构造参数为args... std::allocator::construct(_data + _tail, std::forward(args)...); //更新尾部元素位置 _tail = (_tail + 1) % _max_size; return true; }  //push 实现两个版本，一个接受左值引用，一个接受右值引用 //接受左值引用版本 bool push(const T& val) { std::cout << "called push const T& version" << std::endl; return emplace(val); }  //接受右值引用版本，当然也可以接受左值引用，T&&为万能引用 // 但是因为我们实现了const T& bool push(T&& val) { std::cout << "called push T&& version" << std::endl; return emplace(std::move(val)); }  //出队函数 bool pop(T& val) { std::lock_guard<std::mutex> lock(_mtx); //判断头部和尾部指针是否重合，如果重合则队列为空 if (_head == _tail) { std::cout << "circular que empty ! " << std::endl; return false; } //取出头部指针指向的数据 // 因为右值引用可以隐式转换为左值引用，所以可以将一个右值引用赋值给左值引用 val = std::move(_data[_head]); //更新头部指针 _head = (_head + 1) % _max_size; return true; }private: size_t _max_size; T* _data; std::mutex _mtx; size_t _head = 0; size_t _tail = 0;};

默认构造函数中，_data(std::allocator::allocate(_max_size))用于为 _data 指针分配一块内存，这块内存可以存储 _max_size 个 T 类型的对象，而_data也是T类型的指针，这是内存分配器类模板std::allocator实现的。

我们在创建环形队列设置的最大长度为Cap，但是在构造函数中，分配给 _data 指针的内存其实是Cap + 1，这是为了区分队列为空和队列满的状态，设计中通常会保留一个额外的空间：

• 空队列：当 head == tail 时，表示队列为空。

• 满队列：当 (tail + 1) % max_size == head 时，表示队列已满。

如果不预留额外空间，那么当 head == tail 时，可能既表示队列为空，也可能表示队列已满，这会导致无法区分这两种状态。举例说明：

假设 Cap = 5，那么数组大小为 max_size = Cap + 1 = 6。状态如下：

初始状态（空队列）

	bash [_, _, _, _, _, _]
 head = 0
 tail = 0

队列添加 1 个元素（满队列）

	mathematica [A, _, _, _, _, _]
 head = 0
 tail = 1

队列添加 5 个元素（满队列）

	mathematica [A, B, C, D, E, _]
 head = 0
 tail = 5

此时，(tail + 1) % max_size == head，表示队列已满。

队列删除 1 个元素

	mathematica [_, B, C, D, E, _]
 head = 1
 tail = 5

此时，head != tail，队列不为空。

若尾指针在队尾（5），当删除一个元素再加入一个元素时，尾指针会重置来到 0，此时（0 + 1）% 6 == 1，满队列。

此外，需要说的是析构函数：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

~CircularQueLk() { //循环销毁 std::lock_guard<std::mutex>  lock(_mtx); //调用内部元素的析构函数 while (_head != _tail) { std::allocator::destroy(_data + _head); _head = (_head+1)%_max_size; } //调用回收操作 std::allocator::deallocate(_data, _max_size);}

std::allocator的destroy方法用于调用指向的元素的析构函数，这里通过while函数调用队列中所有元素的析构函数（如果T是基本类型比如 int，那么销毁操作不会有实际效果）；

std::allocator的deallocate方法用于释放通过std::allocator::allocate分配的内存块。这仅回收内存，不会调用元素的析构函数，因此需要先在循环中显式销毁每个元素。

最后需要注意的一点是，再pop函数中，有这么一行代码：val = std::move(_data[_head])，其中，val 是一个T&类型的变量，而std::move返回的类型其实是一个右值引用，我们可以将右值引用赋值给一个左值引用，因为右值引用可以隐式转换为左值引用。但我们不能将一个右值赋值给一个左值引用，那是不合法的。

3.2 实现无锁环形队列（有缺陷）

接下来我们通过原子类型以及内存次序取代其他同步方法实现线程安全的环形队列，该队列是无锁并发的。代码如下：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

template<typename T, size_t Cap>class CircularQueSeq :private std::allocator{public: // 默认构造函数，为 _data 指针分配能容纳 _max_size 个 _data 类型的连续内存块 CircularQueSeq() :_max_size(Cap + 1), _data(std::allocator::allocate(_max_size)), _atomic_using(false),_head(0), _tail(0) {} CircularQueSeq(const CircularQueSeq&) = delete; CircularQueSeq& operator = (const CircularQueSeq&) volatile = delete; CircularQueSeq& operator = (const CircularQueSeq&) = delete;  ~CircularQueSeq() { //循环销毁 bool use_expected = false; bool use_desired = true; do { use_expected = false; use_desired = true; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); //调用内部元素的析构函数 while (_head != _tail) { std::allocator::destroy(_data + _head); _head = （_head+1）% _max_size; } //调用回收操作 std::allocator::deallocate(_data, _max_size); do { use_expected = true; use_desired = false; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); }  //先实现一个可变参数列表版本的插入函数最为基准函数 template <typename ...Args> bool emplace(Args && ... args) { bool use_expected = false; bool use_desired = true; do { use_expected = false; use_desired = true; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); //判断队列是否满了 if ((_tail + 1) % _max_size == _head) { std::cout << "circular que full ! " << std::endl; do { use_expected = true; use_desired = false; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); return false; } //在尾部位置构造一个T类型的对象，构造参数为args... std::allocator::construct(_data + _tail, std::forward(args)...); //更新尾部元素位置 _tail = (_tail + 1) % _max_size; do { use_expected = true; use_desired = false; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); return true; }  //push 实现两个版本，一个接受左值引用，一个接受右值引用 //接受左值引用版本 bool push(const T& val) { std::cout << "called push const T& version" << std::endl; return emplace(val); }  //接受右值引用版本，当然也可以接受左值引用，T&&为万能引用 // 但是因为我们实现了const T& bool push(T&& val) { std::cout << "called push T&& version" << std::endl; return emplace(std::move(val)); }  //出队函数 bool pop(T& val) { bool use_expected = false; bool use_desired = true; do { use_desired = true; use_expected = false; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); //判断头部和尾部指针是否重合，如果重合则队列为空 if (_head == _tail) { std::cout << "circular que empty ! " << std::endl; do { use_expected = true; use_desired = false; } while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); return false; } //取出头部指针指向的数据 val = std::move(_data[_head]); //更新头部指针 _head = (_head + 1) % _max_size; do { use_expected = true; use_desired = false; }while (!_atomic_using.compare_exchange_strong(use_expected, use_desired)); return true; }private: size_t _max_size; T* _data; std::atomic<bool> _atomic_using; // 使用原子变量代替互斥 size_t _head = 0; size_t _tail = 0;};

实现过程其实大差不差，只不过使用原子操作将使用锁的部分代替，而且相比锁的实现，无锁代码更加复杂一些。在这里，我们使用类型为std::atomic＜bool>的变量代替了有锁版本的的成员变量std::mutex，这是为了使用自旋锁的思路将锁替换为原子变量循环检测的方式，接下来分析一下需要关注的成员函数。

a. 析构函数

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

bool use_expected = false;bool use_desired = true;do{ use_expected = false; use_desired = true;}while (!_atomic_using.compare_exchange_strong(use_expected, use_desired));

第一个循环通过将标志位 _atomic_using 置为true确保当前线程独占，防止多个线程同时销毁资源。

_atomic_using 在构造时被初始化为false，所以使用第一个do-while时，会将_atomic_using 置为true，表示当前线程独占，只有当前线程可以销毁资源。

第一个循环执行完后，开始销毁资源，步骤和有锁环形队列相同，就不再过多叙述。

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

do{ use_expected = true; use_desired = false;}while (!_atomic_using.compare_exchange_strong(use_expected, use_desired));

当执行完资源销毁步骤后，执行第二个do-while循环，将_atomic_using置为false，表示当前线程释放对 _atomic_using 的独占访问权，将其设置为未使用状态。

b. 其他成员函数

其他成员函数中，也使用第一个循环和第二个循环代替锁，实现同步机制，就不继续说明了。只需记住，第一个do-while循环相当于加锁，第二个do-while循环相当于解锁，可以理解为是一个没有RAII回收机制的unique_ptr。

3.3 实现无锁环形队列（无缺陷）

虽然通过单个原子变量实现了一个线程安全的环形队列，但是也有弊端：

因为仅有一个线程能独占atomic_using，所有多个线程执行相同的操作时，比如pop，有且仅有一个线程可以获得atomic_using的独占权从而执行，而其他线程会陷入终而复始的等待中。而循环无疑是对CPU资源的浪费，可能会造成其他线程的“受饿”情况，即某个线程被执行无锁操作的线程抢占CPU资源（频繁的自旋重试会造成CPU资源的浪费），自身只分配到极少的执行时间，甚至完全没有，运行几乎停滞或完全停滞。

所以我们可以考虑使用多个原子变量将上述操作优化：

在环形队列的多线程使用中，写入数据的关键在于指针的移动，而不是数据本身的写入。由于不同线程写入的数据位置由指针决定，只要指针的更新是安全的，各线程写入的内存区域就不会冲突。因此，写入操作可以并发进行，无需额外保护。我们只需通过原子操作确保指针的加减是安全的，避免多线程竞争导致状态不一致。这样，数据写入过程是独立的，而指针的原子更新则保证了队列操作的整体正确性和线程安全性。

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

 CircularQueLight():_max_size(Cap + 1), _data(std::allocator::allocate(_max_size)), _head(0), _tail(0) {}private: size_t _max_size; T* _data; std::atomic<size_t>  _head; std::atomic<size_t> _tail;

将无锁版本的私有成员变量修改为上述四个，无需使用_atomic_using来模仿自旋锁的操作，直接将头指针和尾指针的类型换为原子类型，我们只需原子操作确保指针的加减是安全的即可。

3.3.1 pop函数

我们先实现简单的pop：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

// 线程安全的pop实现bool pop(T& val) { size_t h; do { h = _head.load(); //1 处 //判断头部和尾部指针是否重合，如果重合则队列为空 if(h == _tail.load()) { return false; } val = _data[h]; // 2处 } while (!_head.compare_exchange_strong(h,  (h+1)% _max_size)); //3 处 return true;}

在pop函数中，我们在 1 处load获取头部head的值，在 2 处采用了复制的方式将头部元素取出赋值给val，而不是通过std::move，因为多个线程同时pop最后只有一个线程成功执行 3 处代码退出，而失败的则需要继续循环，从更新后的head处pop元素。所以不能用std::move，否则会破坏原有的队列数据。最后，判断当前线程持有的h值和头指针是否相同，如果相同则+1，反之重新循环pop。可能不好理解，我这里详细解释一下：

为什么不能使用 std::move？

在 pop 函数中，多个线程可能同时尝试从队列中弹出元素（而在锁或者自旋锁的保护下，仅有一个线程pop），但最终只有一个线程能够成功更新_head指针。对于未成功更新指针的线程，它们需要重新获取最新的_head值，并从新的位置继续尝试弹出。

如果在2 处使用std::move，会将队列中当前_head指针指向位置的数据转移（move）到val中，这会破坏队列中该位置的数据。结果是，当其他线程在失败后重新尝试弹出时，该位置的数据可能已经被破坏（变为空的、无效的状态），导致数据丢失或逻辑错误。

为什么最终只有一个线程成功？

弹出操作依赖于 compare_exchange_strong 来更新 _head 指针，而这是一个原子操作：

• 只有当 _head 的当前值等于期望值（即线程读取的 h）时，才能成功将 _head 更新为新值。

• 如果某个线程在尝试更新 _head 时，发现 _head 已经被其他线程更新，则说明该线程失败，必须重新尝试。

这意味着，在并发环境下，尽管多个线程可以同时尝试 pop，最终只有一个线程能成功更新 _head 并退出循环，其他线程必须重新获取新的 _head 并继续尝试。

3.3.2 push函数

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

// 存在线程安全的 push 实现bool push(T& val){ size_t t; do { t = _tail.load(); //1 //判断队列是否满 if( (t+1)%_max_size == _head.load()) { return false; } _data[t] = val; //2 } while (!_tail.compare_exchange_strong(t, (t + 1) % _max_size)); //3 return true;}

在 push 函数中，逻辑和pop函数差不多，都是多个线程可能同时push数据，但最终只有一个线程能push进入，而其他线程重新循环重新push。过程虽然差不多，但是push的实现其实存在线程安全问题：

比如线程1 push(1) 而线程2 push(2)，很有可能的顺序是，线程1走到了 2 处将data[t]成功写入了1，线程2晚一点走到了 2 处将data[t]修改为了2， 因为两个线程是同时执行的，所以此时尾指针的值还未被修改，如果线程1先一步修改尾指针，虽然能成功修改，但是内存中的值并不是线程1想要的1，而是2。流程为：1.1 -> 1.2 -> 2.1 -> 2.2 -> 1.3

这样我们看到的效果就是_data[t]被存储为2了，而实际情况应该是被存储为1，因为线程1的原子变量生效，而线程2的原子变量不满足需继续循环。我们需要想办法把_data[t]修改为1，重新优化push函数：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

bool push(T& val){ size_t t; do { t = _tail.load(); //1 //判断队列是否满 if( (t+1)%_max_size == _head.load()) { return false; } } while (!_tail.compare_exchange_strong(t, (t + 1) % _max_size)); //3 _data[t] = val; //2 return true;}

在该版本push函数中，我们先更新指针然后再修改内容，这样能保证多个线程push，仅有一个线程生效时，它写入的数据一定是本线程要写入到tail的数据，而此时tail被缓存在t里，那是一个线程本地变量，所以在这种情况下我们能确定即使多个线程运行到2处，他们的t值也是不同的，并不会产生上面所说的线程安全问题。

但是这种push操作仍然会有其他安全问题：

因为我们是先修改指针，后修改内存的内容，但如果我们更新完指针，在执行 2 处写操作未完成的时候，其他线程调用了pop函数，那么此时读到的值并不是更新后的值（写操作还未完成），而是该片内存原本的值。

我们理解中的同步应该是读操作能读到写操作更新后的值，而不是更新前的值，我们可以增加一个原子变量_tail_update来标记尾部数据是否修改完毕，如果没有修改完毕，此时其他线程pop获取的数据是不安全的，pop返回false。

3.3.3 优化后的pop和push函数

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

bool push(const T& val){ size_t t; do { t = _tail.load(); //1 //判断队列是否满 if( (t+1)%_max_size == _head.load()) { return false; } } while (!_tail.compare_exchange_strong(t, (t + 1) % _max_size)); //3 _data[t] = val; //2 // 数据成功写入之后更新tailup的值 size_t tailup; do { tailup = t; } while (_tail_update.compare_exchange_strong(tailup,  (tailup + 1) % _max_size)); return true;} bool pop(T& val) { size_t h; do { h = _head.load(); //1 处 //判断头部和尾部指针是否重合，如果重合则队列为空 if(h == _tail.load()) { return false; } //判断如果此时要读取的数据和tail_update是否一致，如果一致说明尾部数据未更新完 if(h == _tail_update.load()) { return false; } val = _data[h]; // 2处 } while (!_head.compare_exchange_strong(h,  (h+1)% _max_size)); //3 处 return true;}

因为当前线程执行pop和push获得的h和t都是一个固定值不会改变，改变的只是head指针和tail指针，所以当数据成功写入后，我们可以在push函数中增加一个do-while循环更新tail_update的值（将tail_update指向tail更新后的位置），表示指向已完成写入的最新位置。

而在pop函数中，如果 pop 发现 _head 与 _tail_update 相同_tail_update仍然指向tail指针的上一个位置（数据刚开始存储时，首尾指针均为0），还没有更新，说明此位置的数据尚未写入完成，因此数据是不安全的，pop 应返回 false。

我们模拟一下二者的执行流程：

在 push 中：

• _tail 先移动，表示分配位置。

• 数据写入完成后，再更新 _tail_update，标记此位置的数据可用。

在 pop 中：

• 检查 _tail_update，如果 _head == _tail_update，说明当前位置的数据尚未写入完成，pop 返回 false。

• 只有 _tail_update 超过 _head 时，才能安全读取队列数据。

我们学习了内存序之后知道，原子操作的默认内存序是先后一致次序memory_order_seq_cst，它能保证所有线程对变量操作的顺序观察一致，但是性能消耗过大，我们可以将先后一致内存模型替换为其他内存序，pop函数的实现如下：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

bool pop(T& val) { size_t h; do { h = _head.load(std::memory_order_relaxed); //1 处 //判断头部和尾部指针是否重合，如果重合则队列为空 if (h == _tail.load(std::memory_order_acquire)) //2处 { std::cout << "circular que empty ! " << std::endl; return false; } //判断如果此时要读取的数据和tail_update是否一致，如果一致说明尾部数据未更新完 if (h == _tail_update.load(std::memory_order_acquire)) //3处 { return false; }  val = _data[h];   } while (!_head.compare_exchange_strong(h, (h + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //4 处 std::cout << "pop data success, data is " << val << std::endl; return true;}

1 处，使用了 memory_order_relaxed，这是因为对于 head 指针的加载，我们并不关心线程之间是否有同步需求，除了需要读取最新的 head 值。这里的目的是获取队列头部的索引，以便判断队列是否为空以及获取数据。由于 memory_order_relaxed 不强制同步，所以多个线程并不会相互等待，也不需要保证加载的 head 值和其他操作的顺序关系。这里使用 relaxed 只是为了提高效率，因为队列中有可能会多次重试。

2 处，当从队列中取数据时，需要保证 head 和 tail 指针的同步性。为了确保在读取队列头部元素之前，tail 指针已经正确更新，我们需要使用 memory_order_acquire。这个内存顺序会使得当前线程等待之前的操作完成，从而确保 tail 指针在当前线程读取之前是最新的。

3 处，再次使用 memory_order_acquire 来确保尾部数据的更新已经完成。通过检查 tail_update，你可以确保队列的尾部元素已完全更新并可供当前线程读取。这里的同步逻辑与 _tail` 相同，确保队列的状态对其他线程是正确同步的。如果尾部尚未更新，当前线程将继续重试，确保不会读取到不一致的状态。

4 处， 使用了两个内存顺序：memory_order_release 和memory_order_relaxed。这是因为 compare_exchange_strong 涉及到读改写，可以使用两种内存序：

• memory_order_release 用于确保在更新 head 指针之前，所有对队列的写操作（如 val = _data[h]）对其他线程可见。这保证了在 head 更新之后，其他线程会看到正确的数据。

• memory_order_relaxed 用于在比较失败时，提升效率，因为在期望条件不匹配时无需进行同步。此时，当前线程会重试，依然不需要等待其他线程完成工作，因此使用 relaxed 来减少同步开销。

push 函数的实现如下：

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

bool push(const T& val){ size_t t; do { t = _tail.load(std::memory_order_relaxed); //1 //判断队列是否满 if ((t + 1) % _max_size == _head.load(std::memory_order_acquire)) // 2 { std::cout << "circular que full ! " << std::endl; return false; } } while (!_tail.compare_exchange_strong(t, (t + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //3  _data[t] = val;  size_t tailup;  do { tailup = t; } while (_tail_update.compare_exchange_strong(tailup, (tailup + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //4 std::cout << "called push data success " << val << std::endl; return true;}

1 处，读取该数据时不需要进行线程的同步，所以使用最节省资源的memory_order_relaxed内存序。

2 处，使用 memory_order_acquire 加载 head 指针，确保在进行满队列检查时，头部指针已经同步更新。

3处，使用compare_exchange_strong来尝试更新尾部指针tail。如果tail指针未被其他线程修改，当前线程会成功更新tail指针并进入push操作。如果tail指针已经被其他线程修改，当前线程会重新读取新的tail值，并继续尝试更新。

• memory_order_release: 这个内存顺序保证了在更新 tail 之前，当前线程对队列的修改对其他线程是可见的。

• memory_order_relaxed: 如果 compare_exchange_strong 操作失败，即尾部指针的预期值与实际值不符，那么当前线程会重试。这时，使用relaxed可以避免同步操作的开销，减少不必要的内存屏障。

4 处， _tail_update的更新同样使用了memory_order_release和memory_order_relaxed内存序，理由同上。

3.3.4 完整代码

	
	

		
		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

		

			
		

	

#pragma once#include #include  template<typename T, size_t Cap>class CircularQueSync : private std::allocator{public: CircularQueSync() :_max_size(Cap + 1), _data(std::allocator::allocate(_max_size)) , _head(0), _tail(0), _tail_update(0) {}  CircularQueSync(const CircularQueSync&) = delete; CircularQueSync& operator = (const CircularQueSync&) volatile = delete; CircularQueSync& operator = (const CircularQueSync&) = delete;  ~CircularQueSync() { //调用内部元素的析构函数 while (_head != _tail) { std::allocator::destroy(_data + _head); _head = (++_head)%_max_size; } //调用回收操作 std::allocator::deallocate(_data, _max_size); }  //出队函数 bool pop(T& val) { size_t h; do { h = _head.load(std::memory_order_relaxed); //1 处 //判断头部和尾部指针是否重合，如果重合则队列为空 if (h == _tail.load(std::memory_order_acquire)) //2处 { std::cout << "circular que empty ! " << std::endl; return false; } //判断如果此时要读取的数据和tail_update是否一致，如果一致说明尾部数据未更新完 if (h == _tail_update.load(std::memory_order_acquire)) //3处 { return false; }  val = _data[h];   } while (!_head.compare_exchange_strong(h, (h + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //4 处 std::cout << "pop data success, data is " << val << std::endl; return true; }  bool push(const T& val){ size_t t; do { t = _tail.load(std::memory_order_relaxed); //1 //判断队列是否满 if ((t + 1) % _max_size == _head.load(std::memory_order_acquire)) // 2 { std::cout << "circular que full ! " << std::endl; return false; } } while (!_tail.compare_exchange_strong(t, (t + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //3  _data[t] = val;  size_t tailup;  do { tailup = t; } while (_tail_update.compare_exchange_strong(tailup, (tailup + 1) % _max_size, std::memory_order_release, std::memory_order_relaxed)); //4 std::cout << "called push data success " << val << std::endl; return true; } private: size_t _max_size; T* _data; std::atomic<size_t>  _head; std::atomic<size_t> _tail; std::atomic<size_t> _tail_update;};

4.无锁环形并发队列的优缺点

优点：

• 由于使用了原子操作和自旋重试机制，这种设计避免了传统的锁机制，因此能够实现高并发。每个线程在修改队列指针时（如 push 或 pop）不会进行阻塞等待，而是通过原子操作保证数据一致性。

• 自旋重试：在 push 或 pop 操作中，如果指针未能成功更新（例如，因为另一个线程修改了指针），线程会重试直到成功。这种方式在并发较低的情况下非常高效，但对于高并发的场景可能会带来额外的开销。

• 操作独立性：push 和 pop 操作是独立的，它们之间没有冲突。因此，push 与 pop 操作可以并发执行，互不干扰。只有当多个线程同时进行 push 或 pop 时，才可能导致自旋重试。

• 与传统的锁机制相比（如互斥锁），无锁机制通过原子操作和内存模型的控制来保证并发访问时的线程安全，而不需要通过上下文切换或阻塞来管理线程。这样可以避免锁竞争带来的性能下降。

缺点：

• 当队列存储的是类对象时，多个push线程可能只有一个线程会成功插入数据，而其他线程则会因为重试而浪费时间。这是因为每次重试时，push线程仍然会尝试拷贝类对象到队列中，而拷贝构造函数的调用会增加开销。尤其是当类对象比较复杂时，这种重复的拷贝开销可能会对性能造成显著影响。所以我们一般使用该方式存储标量而不应该存储类对象。

• 如果多个线程频繁并发进行 push 操作，重试机制可能导致每个线程都反复读取、判断和更新队列指针，这样虽然能够保证数据一致性，但会消耗大量 CPU 资源。尤其在高并发情况下，如果队列的插入操作频繁失败并重试，这种开销可能会成为瓶颈。所以我们应该尽量让push和pop并发，而不是多线程并发push。

为什么当任务执行时间比较长的时候，不适合用无锁队列？

无锁队列通常通过原子操作来保证线程安全，在并发环境中保证数据的一致性。但是，原子操作通常是在忙等待（自旋）模式下执行的。当任务执行时间较长时，如果线程长时间占用 CPU 资源进行无锁操作，它可能会导致其他线程的性能下降，甚至引发资源争用。尤其是在任务复杂且需要较多计算的场景下，长时间自旋会导致系统负载过重，影响整个系统的响应性。

因为原子操作相当于自旋重试，如果无锁操作执行时间过长，有可能会导致某一个线程处于“受饿”状态，即某个线程被执行无锁操作的线程抢占CPU资源（频繁的自旋重试会造成CPU资源的浪费），自身只分配到极少的执行时间，甚至完全没有，运行几乎停滞或完全停滞。

无锁队列在短时间、高并发、低延迟的任务场景下表现优秀，但在任务执行时间较长的情况下，使用无锁队列会导致 CPU 资源浪费、过度的自旋等待以及频繁的上下文切换。对于长时间执行的任务，使用带锁的队列是更合适的选择，因为它能有效避免这些问题。

在无锁队列中，当线程在等待队列操作完成时，如果操作需要较长时间处理，线程可能会一直进行自旋等待（即循环尝试获取队列操作的锁）。如果任务执行时间较长，线程就会频繁地进行自旋，导致 CPU 资源的浪费。相反，如果使用带锁或者条件变量的队列，线程可以在等待时挂起进入阻塞状态，释放 CPU 资源，其他线程可以继续运行。