当前位置：首页 > news >正文

【CPP】单生产者单消费者无锁队列使用记录

news 2025/11/24 16:38:11

无锁队列地址：https://github.com/cameron314/readerwriterqueue

该仓库提供三种队列：

无锁队列
带阻塞与超时的无锁队列
无锁环形缓存
以下通过三个官方例子与简要说明进行阐述。

1. 无锁队列

1.1 打印输出函数

#include <readerwriterqueue.h>
#include <iostream>template <class T>
void PrintQueue(moodycamel::ReaderWriterQueue<T>& queue, const char* description = nullptr) {std::cout << "------------------------------------------------ " << std::endl;std::cout << "description  : " << description << std::endl;std::cout << "max_capacity : " << queue.max_capacity() << std::endl;std::cout << "size_approx  : " << queue.size_approx() << std::endl;std::cout << "peek address : " << queue.peek() << std::endl;T    element;bool succeeded;while (queue.size_approx() > 0) {succeeded = queue.try_dequeue(element);if (succeeded) {std::cout << element << " ";}}std::cout << std::endl;
}

1.2 查看基本信息

{// 默认构造moodycamel::ReaderWriterQueue<int> q;  // max_capacity: 15PrintQueue(q, "default");// 有参构造，查看最大容量moodycamel::ReaderWriterQueue<int> q3(3);  // max_capacity: 3PrintQueue(q3, "3");// 有参构造，查看最大容量moodycamel::ReaderWriterQueue<int> q5(5);  // max_capacity: 7PrintQueue(q5, "5");
}

输出：

------------------------------------------------ 
description  : default
max_capacity : 15
size_approx  : 0
peek address : 0------------------------------------------------ 
description  : 3
max_capacity : 3
size_approx  : 0
peek address : 0------------------------------------------------ 
description  : 5
max_capacity : 7
size_approx  : 0
peek address : 0

注意
如果给定的容量大小为2^n -1，则容量为指定值；否则，容量为大于给定值且最小的2^n-1

1.3 插入元素

1.3.1 enqueue/emplace

{moodycamel::ReaderWriterQueue<int> q(3);PrintQueue(q, "init");for (int index = 0; index < 5; index++) {// q.enqueue(index);q.emplace(index);}PrintQueue(q, "enqueue/emplace");  // 0 1 2 3 4 5 6 7
}

输出：

------------------------------------------------ 
description  : init
max_capacity : 3
size_approx  : 0
peek address : 0------------------------------------------------ 
description  : enqueue/emplace
max_capacity : 10
size_approx  : 5
peek address : 0x7e5ca0
0 1 2 3 4

注意
函数enqueue/emplace在队列满后，会分配新的空间以存放插入元素。队列的容量会发生变化。这里仅验证了二次分配的规律：

指定大小容量大小二次分配容量大小
3 3 10
5 7 22
9 15 46

二次分配的容量为首次分配容量的三倍减一，即3n-1

指定大小	容量大小	二次分配容量大小
3	3	10
5	7	22
9	15	46

1.3.2 try_enqueue/try_emplace

{moodycamel::ReaderWriterQueue<int> q(5);for (int index = 0; index < 10; index++) {// q.try_emplace(index);if (!q.try_enqueue(index)) {std::cout << "Insert element is failed, element: " << index << std::endl;}}PrintQueue(q, "try_enqueue/try_emplace");  // 0 1 2 3 4 5 6
}

输出

Insert element is failed, element: 7
Insert element is failed, element: 8
Insert element is failed, element: 9
------------------------------------------------ 
description  : try_enqueue/try_emplace
max_capacity : 7
size_approx  : 7
peek address : 0x725ca0
0 1 2 3 4 5 6

注意
函数try_enqueue/try_emplace在队列满时，不会分配新的空间以存储插入元素。插入元素被丢弃。

1.4 完整代码

#include <readerwriterqueue.h>#include <iostream>template <class T>
void PrintQueue(moodycamel::ReaderWriterQueue<T>& queue, const char* description = nullptr) {std::cout << "------------------------------------------------ " << std::endl;std::cout << "description  : " << description << std::endl;std::cout << "max_capacity : " << queue.max_capacity() << std::endl;std::cout << "size_approx  : " << queue.size_approx() << std::endl;std::cout << "peek address : " << queue.peek() << std::endl;T    element;bool succeeded;while (queue.size_approx() > 0) {succeeded = queue.try_dequeue(element);if (succeeded) {std::cout << element << " ";}}std::cout << std::endl;
}int main() {{// 默认构造moodycamel::ReaderWriterQueue<int> q;  // max_capacity: 15PrintQueue(q, "default");// 有参构造，查看最大容量moodycamel::ReaderWriterQueue<int> q3(3);  // max_capacity: 3PrintQueue(q3, "3");// 有参构造，查看最大容量moodycamel::ReaderWriterQueue<int> q5(5);  // max_capacity: 7PrintQueue(q5, "5");}{moodycamel::ReaderWriterQueue<int> q(5);for (int index = 0; index < 10; index++) {// q.try_emplace(index);if (!q.try_enqueue(index)) {std::cout << "Insert element is failed, element: " << index << std::endl;}}PrintQueue(q, "try_enqueue/try_emplace");  // 0 1 2 3 4 5 6}{moodycamel::ReaderWriterQueue<int> q(3);PrintQueue(q, "init");for (int index = 0; index < 5; index++) {// q.enqueue(index);q.emplace(index);}PrintQueue(q, "enqueue/emplace");  // 0 1 2 3 4 5 6 7}return EXIT_SUCCESS;
}

2. 带阻塞与超时的无锁队列

基本使用与无所队列相同，提供了额外的阻塞函数。

#include <readerwriterqueue.h>#include <iostream>template <class T>
void PrintQueue(moodycamel::BlockingReaderWriterQueue<T>& queue, const char* description = nullptr) {std::cout << "------------------------------------------------ " << std::endl;std::cout << "description  : " << description << std::endl;std::cout << "max_capacity : " << queue.max_capacity() << std::endl;std::cout << "size_approx  : " << queue.size_approx() << std::endl;std::cout << "peek address : " << queue.peek() << std::endl;T    element;bool succeeded;while (queue.size_approx() > 0) {succeeded = queue.try_dequeue(element);if (succeeded) {std::cout << element << " ";}}std::cout << std::endl;
}int main() {{moodycamel::BlockingReaderWriterQueue<int> q;  // max_capacity: 15PrintQueue(q, "default");moodycamel::BlockingReaderWriterQueue<int> q3(3);  // max_capacity: 3PrintQueue(q3, "3");moodycamel::BlockingReaderWriterQueue<int> q5(5);  // max_capacity: 7PrintQueue(q5, "5");}{moodycamel::BlockingReaderWriterQueue<int> q(5);for (int index = 0; index < 8; index++) {// q.enqueue(index);q.emplace(index);}PrintQueue(q, "enqueue/emplace");  // 0 1 2 3 4 5 6 7}{moodycamel::BlockingReaderWriterQueue<int> q(5);for (int index = 0; index < 8; index++) {// q.try_emplace(index);if (!q.try_enqueue(index)) {std::cout << "Insert element is failed, element: " << index << std::endl;}}PrintQueue(q, "try_enqueue/try_emplace");  // 0 1 2 3 4 5 6}{moodycamel::BlockingReaderWriterQueue<int> q(5);for (int index = 0; index < 8; index++) {q.emplace(index);}int element1;q.wait_dequeue(element1);  // 如果队列为空，函数会阻塞到队列非空int  element2;if (q.wait_dequeue_timed(element2, 5)) {  // 如果队列为空，函数会阻塞到超时结束，然后返回falsestd::cout << "element2: " << element2 << std::endl;}PrintQueue(q, "wait_dequeue/wait_dequeue_timed");}return EXIT_SUCCESS;
}

3. 无锁环形缓存

#include <readerwritercircularbuffer.h>#include <iostream>template <class T>
void PrintQueue(moodycamel::BlockingReaderWriterCircularBuffer<T>& queue, const char* description = nullptr) {std::cout << "------------------------------------------------ " << std::endl;std::cout << "description  : " << description << std::endl;std::cout << "max_capacity : " << queue.max_capacity() << std::endl;std::cout << "size_approx  : " << queue.size_approx() << std::endl;T    element;bool succeeded;while (queue.size_approx() > 0) {succeeded = queue.try_dequeue(element);if (succeeded) {std::cout << element << " ";}}std::cout << std::endl;
}int main() {{moodycamel::BlockingReaderWriterCircularBuffer<int> q3(3);  // max_capacity: 3PrintQueue(q3, "3");moodycamel::BlockingReaderWriterCircularBuffer<int> q5(5);  // max_capacity: 5PrintQueue(q5, "5");// 容量为指定位置}{moodycamel::BlockingReaderWriterCircularBuffer<int> q(5);for (int index = 0; index < 8; index++) {if (!q.try_enqueue(index)) {  // 如果插入元素时，队列已满，则丢弃std::cout << "Insert element is failed, element: " << index << std::endl;}}PrintQueue(q, "try_enqueue");  // 0 1 2 3 4}{moodycamel::BlockingReaderWriterCircularBuffer<int> q(5);for (int index = 0; index < 8; index++) {if (!q.wait_enqueue_timed(index, 5)) {  // 队列已满且在指定的超时时间内未获得存储空间，则插入失败，元素被丢弃std::cout << "Insert element is failed, element: " << index << std::endl;}}PrintQueue(q, "wait_enqueue_timed");  // 0 1 2 3 4}{moodycamel::BlockingReaderWriterCircularBuffer<int> q(5);for (int index = 0; index < 8; index++) {q.try_enqueue(index);}int element1;q.wait_dequeue(element1);std::cout << "wait_dequeue->element1: " << element1 << std::endl;int  element2;if (q.wait_dequeue_timed(element2, std::chrono::milliseconds(1))) {std::cout << "element2: " << element2 << std::endl;}PrintQueue(q, "wait_dequeue/wait_dequeue_timed");  // 2 3 4}return EXIT_SUCCESS;
}