一、为什么需要Ring Buffer前面几篇已经学习了多种线程安全队列例如mutex deque ↓ condition_variable阻塞队列 ↓ MPSC无锁链表 ↓ 侵入式MPSC ↓ 双链表消息队列这些队列很多都是基于链表实现的。例如Node1 → Node2 → Node3 → Node4 → nullptr每加入一个节点就可能需要new Node();取出节点之后又需要delete node;这种方式最大的好处就是队列可以比较灵活地动态增长。但是在一些高频数据传输场景中我们可能并不希望频繁new ↓ 使用 ↓ delete因为动态内存分配本身存在一定开销。于是可以换一种完全不同的设计提前申请一块固定大小的连续内存然后循环使用这一块内存。例如提前创建int buffer[8];内存结构下标 0 1 2 3 4 5 6 7 ┌────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┘然后使用两个位置read write分别表示read 下一次从哪里读取数据 write 下一次往哪里写入数据例如依次加入10 20 30数组变成0 1 2 3 4 5 6 7 ┌────┬────┬────┬────┬────┬────┬────┬────┐ │ 10 │ 20 │ 30 │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┘ ↑ ↑ read write消费者取走10 20以后0 1 2 3 4 5 6 7 ┌────┬────┬────┬────┬────┬────┬────┬────┐ │旧值│旧值│ 30 │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┘ ↑ ↑ read write这里需要注意数据被取出之后不一定需要把对应内存真的清零。因为真正决定数据是否有效的是read write而不是数组里面看起来还有没有旧数据。继续不断写入40 50 60 70 80最终write会走到数组末尾。那接下来怎么办如果前面的0 1位置已经被消费者使用完了就可以重新回到0继续写。于是整个数组就像首尾连接起来0 ┌───────┐ 7 │ │ 1 │ │ 6 │ │ 2 │ │ 5 │ │ 3 └───────┘ 4这就是Ring Buffer 环形缓冲区虽然物理内存实际上仍然是一段普通数组0 1 2 3 4 5 6 7但是在逻辑上我们把7的下一个位置重新看成0于是形成一个“环”。二、Ring Buffer最核心的两个下标先实现一个最基础的环形队列#include cstddef #include vector templatetypename T class RingBuffer { private: std::vectorT buffer_; // 真正保存数据的数组 size_t capacity_; // 数组容量 size_t read_; // 下一次读取的位置 size_t write_; // 下一次写入的位置 public: explicit RingBuffer(size_t capacity) : buffer_(capacity), capacity_(capacity), read_(0), write_(0) {} };最重要的两个变量就是size_t read_; size_t write_;刚创建的时候read 0 write 0也就是0 1 2 3 4 ┌────┬────┬────┬────┬────┐ │ │ │ │ │ │ └────┴────┴────┴────┴────┘ ↑ read write生产者加入一个元素10以后0 1 2 3 4 ┌────┬────┬────┬────┬────┐ │ 10 │ │ │ │ │ └────┴────┴────┴────┴────┘ ↑ ↑ read write继续加入20变成0 1 2 3 4 ┌────┬────┬────┬────┬────┐ │ 10 │ 20 │ │ │ │ └────┴────┴────┴────┴────┘ ↑ ↑ read write消费者取出一个10那么0 1 2 3 4 ┌────┬────┬────┬────┬────┐ │旧值│ 20 │ │ │ │ └────┴────┴────┴────┴────┘ ↑ ↑ read write所以write不断向前走 ↓ 负责生产数据 read不断向前走 ↓ 负责消费数据整个队列的工作本质上就是维护这两个下标。三、如何让下标走到末尾以后重新回到0Ring Buffer 最关键的问题就是下标到达数组末尾以后怎么重新回到数组开头假设容量capacity_ 5;那么合法下标只有0 1 2 3 4如果当前write_ 4;正常执行write_;就会得到write_ 5;但是buffer_[5];已经越界了。所以我们希望0 → 1 → 2 → 3 → 4 ↑ ↓ └─────────────────┘也就是4的下一个位置重新变成0最常见的实现就是取模write_ (write_ 1) % capacity_;例如capacity 5当前write 0计算(0 1) % 5 1于是0 → 1当前write 3那么(3 1) % 5 4得到3 → 4最关键的是write 4计算(4 1) % 5 0于是4 → 0这样就形成了循环。读取下标也是一样read_ (read_ 1) % capacity_;所以 Ring Buffer 中非常经典的一句代码就是index (index 1) % capacity;它实际上完成了到末尾 ↓ 回到开头 ↓ 继续使用这也是“环形”最核心的实现。四、如何判断队空和队满现在出现一个新的问题。如果read_ write_;到底代表什么刚创建队列read 0 write 0明显表示队列为空但是如果生产者不断写入write绕了一圈重新追上readread write又可能代表队列已经满了于是产生歧义read write 到底是 队空 还是 队满一种非常经典的解决方案就是故意空出一个位置不用。规定read write表示队空而write的下一个位置 read表示队满于是判断队空bool Empty() const { return read_ write_; }判断队满bool Full() const { return (write_ 1) % capacity_ read_; }假设capacity 5现在read 0 write 4计算(write 1) % capacity (4 1) % 5 0发现0 read所以Full true结构相当于0 1 2 3 4 ┌────┬────┬────┬────┬────┐ │空位│ 10 │ 20 │ 30 │ 40 │ └────┴────┴────┴────┴────┘ ↑ ↑ read write这里虽然数组有5个位置但是实际最多保存4个元素也就是实际可用容量 capacity - 1为什么宁愿浪费一个位置因为这样判断非常简单read_ write_ // 空 (write_ 1) % capacity_ read_ // 满在环形队列中这是非常经典的一种实现方式。当然还可以额外维护size_t size_;通过size 0判断空通过size capacity判断满。但是这样就多维护了一个共享状态。理解 Ring Buffer 时先掌握“空一个位置”的方案会更加直观。五、实现Push和Pop有了buffer read write以及Empty Full之后就可以实现真正的入队和出队。1. Push入队bool Push(const T value) { if (Full()) return false; buffer_[write_] value; write_ (write_ 1) % capacity_; return true; }首先if (Full()) return false;如果已经满了就不能继续写。如果没有满buffer_[write_] value;把数据放到当前写位置。然后write_ (write_ 1) % capacity_;写下标向前移动。例如write 2执行Push(100);先buffer[2] 100然后write 3结构0 1 2 3 ┌────┬────┬─────┬────┐ │ │ │ 100 │ │ └────┴────┴─────┴────┘ ↑ ↑ 数据 write2. Pop出队bool Pop(T value) { if (Empty()) return false; value buffer_[read_]; read_ (read_ 1) % capacity_; return true; }首先if (Empty()) return false;如果队列为空就没有数据可以读取。否则value buffer_[read_];取出当前数据。然后read_ (read_ 1) % capacity_;读取位置向前移动。例如0 1 2 3 ┌────┬────┬────┬────┐ │ 10 │ 20 │ 30 │ │ └────┴────┴────┴────┘ ↑ ↑ read write执行Pop(value);得到value 10然后read 1于是0 1 2 3 ┌────┬────┬────┬────┐ │旧值│ 20 │ 30 │ │ └────┴────┴────┴────┘ ↑ ↑ read write这里buffer[0]可能依然保存10但是这个10已经是无效数据。因为read已经移动到了1。所以 Ring Buffer 中真正决定数据有效范围的是读写位置而不是数组中的旧数据有没有被清零。六、加入mutex实现线程安全Ring Buffer前面的 Ring Buffer 本身还不是线程安全的。假设两个生产者同时queue.Push(100); queue.Push(200);它们都会读取和修改write_;就可能产生数据竞争。消费者同样会修改read_;因此最简单的方法仍然是加入std::mutex完整实现如下#include cstddef #include mutex #include vector templatetypename T class RingBuffer { private: std::vectorT buffer_; // 固定大小缓冲区 size_t capacity_; // 缓冲区大小 size_t read_; // 下一次读取位置 size_t write_; // 下一次写入位置 mutable std::mutex mutex_; public: explicit RingBuffer(size_t capacity) : buffer_(capacity 1), capacity_(capacity 1), read_(0), write_(0) {} // 入队 bool Push(const T value) { std::lock_guardstd::mutex lock(mutex_); if ((write_ 1) % capacity_ read_) return false; buffer_[write_] value; write_ (write_ 1) % capacity_; return true; } // 出队 bool Pop(T value) { std::lock_guardstd::mutex lock(mutex_); if (read_ write_) return false; value buffer_[read_]; read_ (read_ 1) % capacity_; return true; } // 判断是否为空 bool Empty() const { std::lock_guardstd::mutex lock(mutex_); return read_ write_; } // 判断是否已满 bool Full() const { std::lock_guardstd::mutex lock(mutex_); return (write_ 1) % capacity_ read_; } };这里构造函数explicit RingBuffer(size_t capacity) : buffer_(capacity 1), capacity_(capacity 1), read_(0), write_(0) {}为什么用户传入capacity以后内部却创建capacity 1因为我们采用了空一个位置来区分队空和队满。例如希望真正保存8个元素那么内部需要9个位置其中一个位置始终用于区分队空 和 队满所以RingBufferint queue(8);表示用户真正可以保存8个int。内部实际上创建buffer_(9);这样使用起来更加符合直觉。七、多线程测试Ring Buffer下面创建生产者和消费者#include atomic #include iostream #include thread int main() { RingBufferint queue(10); std::atomicbool finished{false}; // 生产者 std::thread producer([]() { for (int i 1; i 20; i) { // 队列满时暂时让出CPU之后继续尝试 while (!queue.Push(i)) std::this_thread::yield(); std::cout push : i std::endl; } finished.store(true); }); // 消费者 std::thread consumer([]() { int value; while (true) { if (queue.Pop(value)) { std::cout pop : value std::endl; continue; } if (finished.load()) break; std::this_thread::yield(); } }); producer.join(); consumer.join(); return 0; }整个过程Producer ↓ Push() ↓ write不断向前移动 ↓ ┌─────────────────────────────┐ │ Ring Buffer │ │ │ │ 0 → 1 → 2 → 3 → ... → N │ │ ↑ ↓ │ │ └─────────────────────┘ │ └─────────────────────────────┘ ↓ read不断向前移动 ↓ Pop() ↓ Consumer当write到末尾0 → 1 → 2 → 3 → 4 ↓ 0消费者的read同样0 → 1 → 2 → 3 → 4 ↓ 0两个下标不断在数组中循环移动write ↓ ┌────────────┐ ┌──│ │──┐ │ │ RingBuffer │ │ └──│ │──┘ └────────────┘ ↑ read因此 Ring Buffer 最核心的其实只有四个概念固定数组 read write 循环移动对应代码buffer_[write_] value; write_ (write_ 1) % capacity_;负责生产。而value buffer_[read_]; read_ (read_ 1) % capacity_;负责消费。八、Ring Buffer和链表队列有什么区别学到这里可以把 Ring Buffer 和前面的链表队列放在一起比较。普通链表队列Node → Node → Node → Node → nullptr通常特点是动态创建节点 ↓ 空间可以灵活增长 ↓ 节点在内存中不一定连续Ring Buffer┌────┬────┬────┬────┬────┬────┐ │ 0 │ 1 │ 2 │ 3 │ 4 │ 5 │ └────┴────┴────┴────┴────┴────┘ ↑ ↓ └────────────────────────┘特点是提前申请固定空间 ↓ 重复使用这一块内存 ↓ 不需要每次入队都创建链表节点可以简单对比链表队列 ──────────────────── 空间可以动态增长 节点通过指针连接 经常涉及new/delete 节点内存可能不连续 Ring Buffer ──────────────────── 容量通常提前确定 数组连续存储 通过read/write定位 可以重复利用已有空间Ring Buffer 还有一个很重要的特点连续内存通常具有更好的缓存局部性。CPU 访问buffer[0] buffer[1] buffer[2] buffer[3]这些数据在内存中通常是连续的。相比链表Node1 → Node2 → Node3节点可能分散在不同内存位置。因此 Ring Buffer 经常出现在网络数据收发 音视频数据缓存 日志系统 生产者消费者通信 高性能消息队列 实时数据处理等场景中。例如音频数据不断到达音频采集线程 ↓ Ring Buffer ↓ 音频处理线程或者网络程序Socket接收线程 ↓ Ring Buffer ↓ 协议解析线程都可以利用环形缓冲区进行数据暂存。需要注意的是本篇实现std::mutex mutex_;所以它仍然属于基于锁的线程安全 Ring Buffer。真正高性能的 Ring Buffer 还可以继续通过std::atomic维护生产者和消费者位置从而减少甚至避免传统mutex。例如后续更加复杂的 Ring Queue 会涉及producer_head producer_tail consumer_head consumer_tail并配合compare_exchange处理多个生产者和多个消费者之间的并发竞争。这也是一些高性能网络框架和消息队列中经常采用的设计。总结Ring Buffer 和前面学习的链表队列最大的区别就是它不再不断创建新节点而是提前准备固定数组 ↓ 通过read读取 ↓ 通过write写入 ↓ 到达数组末尾后重新回到0 ↓ 循环使用同一块内存其中最经典的一句代码index (index 1) % capacity;完成的就是0 → 1 → 2 → 3 → ... → N ↑ ↓ └───────────────────────┘而使用“预留一个空位置”的方案后read_ write_表示队列为空(write_ 1) % capacity_ read_表示队列已满。所以一个基础 Ring Buffer 可以概括成数组 read/write下标 取模循环 队空/队满判断如果再加入std::mutex就可以得到一个最基础的线程安全 Ring Buffer。从整个线程安全队列系列来看目前已经学习了一mutex deque线程安全队列 ↓ 二condition_variable生产者消费者队列 ↓ 三atomic实现MPSC无锁队列 ↓ 四侵入式MPSC队列 ↓ 五双链表 双锁消息队列 ↓ 六Ring Buffer环形队列这些实现虽然都叫“队列”但背后的设计思想已经从最简单的给STL容器加一把锁逐渐发展到减少锁竞争 → 原子操作 → 侵入式链表 → 双缓冲思想 → 固定内存循环利用理解这些不同实现方式之后再去看更加复杂的高性能 Ring Queue就会更容易理解为什么里面需要多个生产者/消费者下标以及 CAS 原子操作。0voice · GitHub