Ringbuffer 无锁环形队列-尧图网络科技

RingBuffer（环形缓冲区）：

是一种使用固定大小数组实现的(SPSC)单生产者单消费者、先进先出（FIFO）数据结构。它在逻辑上首尾相连，形成一个环形，可以循环使用内存空间。

RingBuffer 的核心特性：

1.固定大小，容量为2的幂。

2.循环利用，写满后从头开始写入、覆盖旧数据（或拒绝写入）。

3.高效的索引计算，使用位运算取模

4.无锁并发：SPSC（单生产者单消费者）可以实现无锁、避免互斥锁的开销。

5.性能优势：预分配内存，无动态分配、操作时间复杂度：O(1) 、缓存友好：连续内存访问

使用场景：

生产者-消费者模式、网络数据包处理、音频/视频流处理、消息队列（低延迟场景）

#include <atomic> #include <cstddef> #include <type_traits> template<typename T, std::size_t Capacity> class RingBuffer { public: static_assert(Capacity && !(Capacity & (Capacity - 1)), "Capacity must be power of 2"); //位运算检查是否为2的幂 RingBuffer() : read_(0), write_(0) {} ~RingBuffer() { std::size_t r = read_.load(std::memory_order_relaxed); std::size_t w = write_.load(std::memory_order_relaxed); while (r != w) { reinterpret_cast<T *>(&buffer_[r])->~T(); r = (r + 1) & (Capacity - 1); } } // 这里使用万能引用和完美转发，支持左值和右值 template<typename U> bool Push(U && value) { const std::size_t w = write_.load(std::memory_order_relaxed); const std::size_t next_w = (w + 1) & (Capacity - 1); // 检查缓冲区是否满 if (next_w == read_.load(std::memory_order_acquire)) { return false; } new (&buffer_[w]) T(std::forward<U>(value)); write_.store(next_w, std::memory_order_release); return true; } bool Pop(T & value) { const std::size_t r = read_.load(std::memory_order_relaxed); // 检查缓冲区是否空 if (r == write_.load(std::memory_order_acquire)) { return false; } // 取出元素并析构 value = std::move(*reinterpret_cast<T *>(&buffer_[r])); reinterpret_cast<T *>(&buffer_[r])->~T(); read_.store((r + 1) & (Capacity - 1), std::memory_order_release); return true; } std::size_t Size() const { const std::size_t r = read_.load(std::memory_order_acquire); const std::size_t w = write_.load(std::memory_order_acquire); return (w >= r) ? (w - r) : (Capacity - r + w); } private: //cache line 64B alignas(64) std::atomic<std::size_t> read_; alignas(64) std::atomic<std::size_t> write_; alignas(64) std::aligned_storage_t<sizeof(T), alignof(T)> buffer_[Capacity]; };

alignas(64) std::atomic<std::size_t> read_; alignas(64) std::atomic<std::size_t> write_; alignas(64) std::aligned_storage_t<sizeof(T), alignof(T)> buffer_[Capacity];

两个原子变量：read_和write_分别由消费者和生产者修改
缓存行对齐：避免伪共享
原始内存存储：支持非POD类型

static_assert(Capacity && !(Capacity & (Capacity - 1)), "Capacity must be power of 2");

这行static_assert用于编译时检查Capacity是否是 2的幂（如 1, 2, 4, 8, 16, 32, 64...）。

static_assert是 C++11 引入的编译时断言，用于在编译期间检查条件是否满足。如果条件为false，编译器会报错并显示自定义错误信息。

内存序的设置：

生产者读取生产者索引 write_ 时用relaxed，生产者读取消费者索引 read_ 时用acquire，生产者修改生产者索引时用release。

if (next_w == read_.load(std::memory_order_acquire)) { return false; } new (&buffer_[w]) T(std::forward<U>(value)); write_.store(next_w, std::memory_order_release); return true;

acquire避免下方代码优化到上方，确保先检查队列是否满再决定下一步。

release避免上方代码优化到下方，确保元素成功插入后再更新write_索引。

消费者读取消费者索引 read_ 时用relaxed，消费者读取生产者索引 write_ 时用acquire，消费者者修改消费者者索引时用release。

if (r == write_.load(std::memory_order_acquire)) { return false; } // 取出元素并析构 value = std::move(*reinterpret_cast<T *>(&buffer_[r])); reinterpret_cast<T *>(&buffer_[r])->~T(); read_.store((r + 1) & (Capacity - 1), std::memory_order_release); return true;

acquire避免下方代码优化到上方，确保先检查队列是否为空再决定下一步。

release避免上方代码优化到下方，确保元素成功取出后再更新read_索引。

为什么要alignas对齐？

alignas(64) std::atomic<std::size_t> read_; alignas(64) std::atomic<std::size_t> write_; alignas(64) std::aligned_storage_t<sizeof(T), alignof(T)> buffer_[Capacity];

缓存行（64字节）是数据传输的最小单位，通过alignas(64)把变量存入不同的缓存行避免伪共享。

如果没有对齐可能会出现：

1. 生产者修改 read_ → 整个缓存行变为"脏" 2. 消费者要修改 write_ → 缓存行失效，重新加载 3. 即使修改的是不同变量，也会互相影响 4. 造成缓存行在核心之间"乒乓"效应，性能急剧下降

如何判断空和满

由于数组是环形的，空状态和满状态看起来一样（都是两个指针指向同一位置），所以必须保留一个空位来区分。

当读指针等于写指针时，表示缓冲区为空，没有数据可读。当写指针的下一个位置等于读指针时，表示缓冲区已满，不能再写入数据。这个"下一个位置"的计算就是索引加一后对容量取模。

因此，容量为 N 的环形缓冲区实际最多只能存储 N 减一个元素。

写入数据的过程

写入数据时，首先读取当前的写指针位置，然后计算下一个写位置，即当前写位置加一后对容量取模。接着检查下一个写位置是否等于当前的读指针，如果相等说明缓冲区已满，写入失败。如果不相等，就在当前写指针指向的位置构造数据对象，最后将写指针更新为刚才计算出的下一个写位置。

读取数据的过程

读取数据时，首先读取当前的读指针位置，然后检查读指针是否等于当前的写指针，如果相等说明缓冲区为空，读取失败。如果不相等，就从读指针指向的位置取出数据，然后析构该位置的对象，最后将读指针更新为下一个位置，即当前读位置加一后对容量取模。

指针如何循环

当指针到达数组最后一个索引时，加一后对容量取模的结果为零，指针就自动回到了数组开头，实现了环形效果。例如容量为八时，索引七的下一个位置是零，索引三的下一个位置是四。

资讯详情