26、环形缓冲区：生产者消费者模型，无锁环形缓冲区的实现

环形缓冲区，这个名字听起来挺唬人。说白了，就是一个固定大小的数组，加上两个指针——一个写，一个读。写指针追着读指针跑，读指针追着写指针跑，形成一个环。嗯，就这么简单。

我在项目中用过很多次环形缓冲区。最典型的就是串口接收数据，中断服务程序往里写，主循环往外读。如果不用环形缓冲区，数据很容易丢。你想想看，中断来得快，主循环处理慢，中间总得有个地方暂存一下。

生产者消费者模型

先理清两个角色：

• 生产者：往缓冲区里写数据。比如中断服务程序、网络接收线程。
• 消费者：从缓冲区里读数据。比如主循环、数据处理线程。

这两个角色跑在不同的执行上下文中。如果处理不好，就会出现数据覆盖、读到脏数据等问题。

无锁环形缓冲区的设计思路

传统做法是用互斥锁保护。但锁的开销不小，尤其是在中断里。我建议用无锁的方式——利用原子操作和内存屏障来保证数据一致性。

无锁环形缓冲区的关键点：

• 写指针：生产者专用，指向下一个要写入的位置。
• 读指针：消费者专用，指向下一个要读取的位置。
• 缓冲区满：写指针追上读指针。
• 缓冲区空：读指针追上写指针。

这里有个细节：如何区分「满」和「空」？两种状态都是指针相等。我常用的做法是——留一个空位。也就是说，缓冲区实际可用大小是 size - 1。当写指针 + 1 等于读指针时，表示满。当写指针等于读指针时，表示空。

代码实现

下面是一个完整的无锁环形缓冲区实现。我习惯用 C11 的 stdatomic.h，这样跨平台性好一些。

#include <stdatomic.h>
#include <stdint.h>
#include <stdbool.h>

#define RING_BUFFER_SIZE 256  // 必须是2的幂

typedef struct {
    uint8_t buffer[RING_BUFFER_SIZE];
    atomic_uint write_index;  // 写指针
    atomic_uint read_index;   // 读指针
} ring_buffer_t;

// 初始化
void ring_buffer_init(ring_buffer_t *rb) {
    atomic_store(&rb->write_index, 0);
    atomic_store(&rb->read_index, 0);
}

// 写入一个字节
bool ring_buffer_put(ring_buffer_t *rb, uint8_t data) {
    unsigned int w = atomic_load(&rb->write_index);
    unsigned int r = atomic_load(&rb->read_index);
    
    // 检查是否满
    if (((w + 1) & (RING_BUFFER_SIZE - 1)) == r) {
        return false;  // 缓冲区满
    }
    
    rb->buffer[w] = data;
    atomic_store(&rb->write_index, (w + 1) & (RING_BUFFER_SIZE - 1));
    return true;
}

// 读取一个字节
bool ring_buffer_get(ring_buffer_t *rb, uint8_t *data) {
    unsigned int w = atomic_load(&rb->write_index);
    unsigned int r = atomic_load(&rb->read_index);
    
    // 检查是否空
    if (w == r) {
        return false;  // 缓冲区空
    }
    
    *data = rb->buffer[r];
    atomic_store(&rb->read_index, (r + 1) & (RING_BUFFER_SIZE - 1));
    return true;
}

// 查询当前数据量
unsigned int ring_buffer_available(ring_buffer_t *rb) {
    unsigned int w = atomic_load(&rb->write_index);
    unsigned int r = atomic_load(&rb->read_index);
    return (w - r) & (RING_BUFFER_SIZE - 1);
}

为什么无锁能工作？

你可能会问：没有锁，不会出问题吗？

关键在于——生产者和消费者操作的是不同的指针。生产者只修改写指针，消费者只修改读指针。这两个指针是独立的原子变量。只要保证每个指针的读写是原子的，就不会出现数据竞争。

我曾经在一个项目中踩过坑。当时用的是普通 int 变量，没有加 atomic。结果在 Cortex-M3 上跑得好好的，换到 Cortex-M0 上就出问题了。为什么？因为 M0 不支持 32 位变量的原子读写，需要多条指令完成。中断来了，指令执行到一半，数据就乱了。

嗯，从那以后，我所有跨平台代码都老老实实用 stdatomic.h。

内存屏障的问题

还有一个容易被忽略的点——内存序。上面的代码用的是默认的 memory_order_seq_cst，这是最严格的顺序一致性。在 x86 上性能还行，但在 ARM 上开销比较大。

如果追求极致性能，可以优化内存序：

• 写数据时，用 memory_order_release 保证之前的数据写入对消费者可见。
• 读数据时，用 memory_order_acquire 保证读到最新的数据。

不过我个人建议，除非你非常清楚自己在做什么，否则先用默认的。性能不够再优化。过早优化是万恶之源，这话不是我说的，是 Knuth 老爷子说的。

SVG 流程图：无锁环形缓冲区核心逻辑

性能对比

我简单测过一组数据，在 Cortex-M4 上跑 100 万次读写：

实现方式	耗时（ms）	代码量（行）
互斥锁 + 环形缓冲区	约 85	约 60
无锁环形缓冲区（默认内存序）	约 32	约 40
无锁环形缓冲区（优化内存序）	约 28	约 45

可以看到，无锁版本比加锁版本快了将近 3 倍。而且代码更简洁，没有锁的初始化、获取、释放这些开销。

避坑指南

我曾经在一个项目中，把缓冲区大小设成了 255。结果位运算取模出了问题。因为 255 不是 2 的幂，(w + 1) & 254 不等于 (w + 1) % 255。嗯，那天排查了整整两个小时。

所以记住：缓冲区大小必须是 2 的幂。这是无锁环形缓冲区的前提条件。

还有一个坑——中断里调用 ring_buffer_put 时，如果缓冲区满了，直接返回 false。这时候不能阻塞等待，因为中断里不能阻塞。我建议的做法是：要么丢弃数据，要么扩大缓冲区。具体怎么选，看你的业务场景。

最后说一句：无锁编程不是银弹。它只适用于特定的场景。如果你的项目逻辑复杂，或者对实时性要求没那么苛刻，用锁反而更安全。别为了炫技而引入不必要的复杂度。

---

公众号:蓝海资料掘金营，微信deep3321