环形缓冲区是嵌入式和流式系统的基石。固定内存、零分配、O(1) 入队出队。但一旦两个线程同时操作它——一个生产、一个消费——你就踏入了并发数据结构的雷区。好消息是:单生产者单消费者(SPSC)场景是唯一可以安全手写的无锁队列。它不需要互斥锁、不需要 CAS 循环,只需要两个原子索引加上正确的内存序。本文从零构建一个 SPSC 环形缓冲区,进行性能基准测试,用 ThreadSanitizer 捕获真实的数据竞态,并与互斥锁保护的队列做对比。
如果你来自裸机 superloop 或 FreeRTOS 背景,你应该已经见过队列的实际应用——不使用 RTOS 的 superloop 和 FreeRTOS 任务与队列 这两篇文章已经覆盖过那些内容。这里我们深入一层:当你只用 C11 原子操作自己实现同步时,队列到底长什么样?
什么是环形缓冲区
环形缓冲区是一个固定大小的数组,逻辑上首尾相连成环。两个索引——head(写位置)和 tail(读位置)——按容量取模递增:
size_t next = (head + 1) & mask; /* mask = 容量 - 1,容量为 2 的幂 */
这就是全部的索引方案。入队时向 head 写入并推进 head;出队时从 tail 读取并推进 tail。没有 malloc,没有 free,没有指针跳转。内存在编译期就确定,这正是微控制器和延迟敏感路径最需要的特性。
head 写入,消费者在 tail 读取。青色槽位存有有效数据;灰色槽位空闲。满/空的歧义
head == tail 可能表示”空”也可能表示”满”。经典解法是牺牲一个槽位:当 (head + 1) % capacity == tail 时视为满。容量为 8 的环形缓冲区实际可用 7 个槽位。另一种方案是使用独立的原子计数器,但每次入队/出队都多一次原子操作。对大多数嵌入式场景来说,牺牲一个槽位是正确的取舍。
SPSC:唯一值得手写的无锁队列
如果使用普通 size_t 索引的朴素环形缓冲区,两个线程一起用就是数据竞态。但在 SPSC 场景下——一个生产者线程、一个消费者线程——同步逻辑可以简化到极致:
- 生产者独占
head(只有它写head),读tail判断是否已满。 - 消费者独占
tail(只有它写tail),读head判断是否为空。 - 每侧只写自己的索引,只读对方的。
这意味着唯一的跨线程通信是:生产者发布 head,消费者读 head;消费者发布 tail,生产者读 tail。两次原子加载、两次原子存储,加上 acquire/release 内存序。无锁、无 CAS、无 ABA 问题。
以下是完整实现:
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
typedef struct {
size_t capacity;
size_t mask;
_Atomic size_t head;
_Atomic size_t tail;
uint64_t *slots;
} spsc_ring_t;
bool spsc_ring_push(spsc_ring_t *r, uint64_t val) {
size_t h = atomic_load_explicit(&r->head, memory_order_relaxed);
size_t next = (h + 1) & r->mask;
if (next == atomic_load_explicit(&r->tail, memory_order_acquire))
return false; /* 满 */
r->slots[h & r->mask] = val;
atomic_store_explicit(&r->head, next, memory_order_release);
return true;
}
bool spsc_ring_pop(spsc_ring_t *r, uint64_t *val) {
size_t t = atomic_load_explicit(&r->tail, memory_order_relaxed);
if (t == atomic_load_explicit(&r->head, memory_order_acquire))
return false; /* 空 */
*val = r->slots[t & r->mask];
atomic_store_explicit(&r->tail, (t + 1) & r->mask, memory_order_release);
return true;
}
二十行代码,就这么简单。
为什么 acquire/release 就够了
内存序的推理是大多数人一带而过的地方。让我们说清楚。
生产者做三件事:(1) 以 acquire 加载 tail,(2) 写入槽位数据,(3) 以 release 存储 head。消费者做镜像操作:(1) 以 acquire 加载 head,(2) 读取槽位数据,(3) 以 release 存储 tail。
对 head 的 release 存储保证:生产者在此之前的全部操作——包括槽位数据的写入——对任何 acquire 加载 head 并看到新值的线程都可见。这就是 C11 内存模型的 synchronizes-with 关系。消费者对 head 的 acquire 加载是对应的边。因此当消费者看到 head 前进时,它保证能看到生产者写入的槽位数据。tail 方向同理。
对自己索引的加载使用 memory_order_relaxed,因为只有一个线程写它——该读取不需要跨线程同步。acquire 只加在对方索引的加载上,release 只加在自己索引的存储上。这就是最小正确的内存序。
真实多线程测试
我用 gcc -O2 -pthread 在本机(GCC 16.2.1)编译上述代码,运行生产者/消费者线程对,通过 4096 槽位的环形缓冲区推送 1000 万条目:
Consumer received 10000000 items (expected 10000000)
First value: 0, last value: 9999999
Produced 10000000 items in 0.163 s
Throughput: 61.21 M ops/sec
Latency: 16 ns/op (amortized, both sides)
一千万人队,一千万人出队,零丢失、零重复。单核每秒 6100 万次操作——这就是无锁、无系统调用、无缓存行弹跳的吞吐量。
OVERWRITE vs DROP 策略
环形缓冲区满时,有两种策略:
- DROP:拒绝新条目。适用于每个样本都重要的场景——传感器读数、网络数据包。
- OVERWRITE:推进
tail丢弃最旧的条目,再写入新数据。适用于只关心最新数据的场景——音频采样、仪表读数、日志消息。
void spsc_ring_push_overwrite(spsc_ring_t *r, uint64_t val) {
size_t h = atomic_load_explicit(&r->head, memory_order_relaxed);
size_t next = (h + 1) & r->mask;
size_t t = atomic_load_explicit(&r->tail, memory_order_acquire);
if (next == t) {
atomic_store_explicit(&r->tail, (t + 1) & r->mask,
memory_order_release);
}
r->slots[h & r->mask] = val;
atomic_store_explicit(&r->head, next, memory_order_release);
}
两种策略,在容量为 8 的环形缓冲区上推送 20 个条目的实测结果:
=== DROP policy (cap=8) ===
Pushed 7, dropped 13
Ring contents (oldest first): 0 1 2 3 4 5 6
=== OVERWRITE policy (cap=8) ===
Pushed 20 items into cap-8 ring with OVERWRITE
Ring contents (oldest first): 13 14 15 16 17 18 19
DROP 保留了前 7 个,拒绝后 13 个。OVERWRITE 保留了最后 7 个,静默丢弃最旧的 13 个。同样的环形缓冲区,同样的代码路径,只多了一个分支——语义完全不同。
ThreadSanitizer 捕获朴素版本的竞态
如果跳过原子操作、使用普通 size_t 索引会怎样?未定义行为——而 ThreadSanitizer 会证明这一点。我写了一个使用普通加载和存储的朴素版本,用 -fsanitize=thread 编译,运行同样的生产者/消费者测试:
==================
WARNING: ThreadSanitizer: data race (pid=1490572)
Read of size 8 at 0x5645fde3f0b0 by thread T2:
#0 naive_ring_pop /tmp/opencode/work/ringbuf/naive_ring.h:47 (test_naive_tsan+0x12f5)
#1 consumer /tmp/opencode/work/ringbuf/test_naive.c:25 (test_naive_tsan+0x12f5)
Previous write of size 8 at 0x5645fde3f0b0 by thread T1:
#0 naive_ring_push /tmp/opencode/work/ringbuf/naive_ring.h:41 (test_naive_tsan+0x125b)
#1 producer /tmp/opencode/work/ringbuf/test_naive.c:15 (test_naive_tsan+0x125b)
Location is global 'ring' of size 40 at 0x5645fde3f0a0
SUMMARY: ThreadSanitizer: data race /tmp/opencode/work/ringbuf/naive_ring.h:47 in naive_ring_pop
==================
Consumer received 100000 items
ThreadSanitizer: reported 2 warnings
两个数据竞态,都在 head/tail 索引上。朴素版本”能跑”——在 x86 上这种时序下,普通存储恰好传播得足够快,输出计数是对的。但这是未定义行为。在弱内存序架构(ARM、Apple Silicon)上,消费者可能看到新的 head 但槽位数据是旧的——一个永远不会在测试中出现、但在生产环境中咬你一口的撕裂读。
然后我用 ThreadSanitizer 编译原子版本的同一测试:
Consumer received 100000 items
干净。零警告。acquire/release 内存序不只是理论——它是”碰巧能跑”和”保证正确”之间的区别。
SPSC 环形缓冲区 vs 互斥锁队列
锁的代价到底有多大?我对 SPSC 环形缓冲区和互斥锁 + 条件变量队列做了基准测试(相同容量、相同条目数、均编译 -O2):
Mutex queue: consumer received 5000000 items
Mutex queue: 1.033 s, 4.84 M ops/sec
SPSC ring: consumer received 5000000 items
SPSC ring: 0.085 s, 59.13 M ops/sec
SPSC is 12.2x faster than mutex queue
快 12 倍。互斥锁版本的时间花在 pthread_mutex_lock/unlock、pthread_cond_wait/signal,以及被争用锁的缓存行弹跳上。SPSC 环形缓冲区每次操作只做两次原子加载和两次原子存储,且都在各自核心的 L1 热缓存行上。在 x86 上,无竞争缓存行上的 LOCK 前缀原子操作与普通加载成本相同——硬件一致性协议完成工作,不需要总线锁。
差距在多核系统和弱内存序架构上更大——互斥锁的内存屏障更昂贵,而 SPSC 环形缓冲区的精准 acquire/release 恰好够用——不强不弱。
什么时候用、什么时候不要用
当你只有一个生产者和一个消费者时使用 SPSC 环形缓冲区:ISR 到线程、UART RX 到解析器、传感器 DMA 到处理任务、音频回调到 DSP。容量为 2 的幂,索引为 atomic_size_t,内存序为 acquire/release。就这样。
不要手写 MPMC(多生产者多消费者)无锁队列。那是一个研究课题——LMAX Disruptor、Vyukov 的 MPMC 队列、并发侵入式链表,这些存在本身就说明问题有多难。写错了意味着丢数据、撕裂读、以及每百万次运行才复现一次的海森堡 bug。用久经考验的实现,或者老老实实用互斥锁。
如果你用 FreeRTOS,记住 xQueueSend/xQueueReceive 已经为你提供了带有正确内存序的 SPSC 安全队列——参见 FreeRTOS 任务与队列 这篇文章。这里的环形缓冲区是当你想理解底层原理、或者在没有 RTOS 开销的情况下追求最低延迟时的选择。
环形缓冲区是那种”二十行 C 写对了就胜过通用锁”的结构。关键在于理解为什么它是正确的:每个索引只有一个写者、跨线程边上的 acquire/release、以及到此为止的自律。