Ringbuffer 无锁环形队列

RingBuffer环形缓冲区是一种使用固定大小数组实现的(SPSC)单生产者单消费者、先进先出FIFO数据结构。它在逻辑上首尾相连形成一个环形可以循环使用内存空间。RingBuffer 的核心特性1.固定大小容量为2的幂。2.循环利用写满后从头开始写入、覆盖旧数据或拒绝写入。3.高效的索引计算使用位运算取模4.无锁并发SPSC单生产者单消费者可以实现无锁、避免互斥锁的开销。5.性能优势预分配内存无动态分配 、操作时间复杂度O(1) 、 缓存友好连续内存访问使用场景生产者-消费者模式、网络数据包处理、音频/视频流处理、消息队列低延迟场景#include atomic #include cstddef #include type_traits templatetypename 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_castT *(buffer_[r])-~T(); r (r 1) (Capacity - 1); } } // 这里使用万能引用和完美转发支持左值和右值 templatetypename 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::forwardU(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_castT *(buffer_[r])); reinterpret_castT *(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::atomicstd::size_t read_; alignas(64) std::atomicstd::size_t write_; alignas(64) std::aligned_storage_tsizeof(T), alignof(T) buffer_[Capacity]; };alignas(64) std::atomicstd::size_t read_; alignas(64) std::atomicstd::size_t write_; alignas(64) std::aligned_storage_tsizeof(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是 C11 引入的编译时断言用于在编译期间检查条件是否满足。如果条件为false编译器会报错并显示自定义错误信息。内存序的设置生产者读取生产者索引 write_ 时用relaxed生产者读取消费者索引 read_ 时用acquire生产者修改生产者索引时用release。if (next_w read_.load(std::memory_order_acquire)) { return false; } new (buffer_[w]) T(std::forwardU(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_castT *(buffer_[r])); reinterpret_castT *(buffer_[r])-~T(); read_.store((r 1) (Capacity - 1), std::memory_order_release); return true;acquire避免下方代码优化到上方确保先检查队列是否为空再决定下一步。release避免上方代码优化到下方确保元素成功取出后再更新read_索引。为什么要alignas对齐alignas(64) std::atomicstd::size_t read_; alignas(64) std::atomicstd::size_t write_; alignas(64) std::aligned_storage_tsizeof(T), alignof(T) buffer_[Capacity];缓存行64字节是数据传输的最小单位通过alignas(64)把变量存入不同的缓存行避免伪共享。如果没有对齐可能会出现1. 生产者修改 read_ → 整个缓存行变为脏 2. 消费者要修改 write_ → 缓存行失效重新加载 3. 即使修改的是不同变量也会互相影响 4. 造成缓存行在核心之间乒乓效应性能急剧下降如何判断空和满由于数组是环形的空状态和满状态看起来一样都是两个指针指向同一位置所以必须保留一个空位来区分。当读指针等于写指针时表示缓冲区为空没有数据可读。当写指针的下一个位置等于读指针时表示缓冲区已满不能再写入数据。这个下一个位置的计算就是索引加一后对容量取模。因此容量为 N 的环形缓冲区实际最多只能存储 N 减一个元素。写入数据的过程写入数据时首先读取当前的写指针位置然后计算下一个写位置即当前写位置加一后对容量取模。接着检查下一个写位置是否等于当前的读指针如果相等说明缓冲区已满写入失败。如果不相等就在当前写指针指向的位置构造数据对象最后将写指针更新为刚才计算出的下一个写位置。读取数据的过程读取数据时首先读取当前的读指针位置然后检查读指针是否等于当前的写指针如果相等说明缓冲区为空读取失败。如果不相等就从读指针指向的位置取出数据然后析构该位置的对象最后将读指针更新为下一个位置即当前读位置加一后对容量取模。指针如何循环当指针到达数组最后一个索引时加一后对容量取模的结果为零指针就自动回到了数组开头实现了环形效果。例如容量为八时索引七的下一个位置是零索引三的下一个位置是四。