嵌入式串口收发的正确姿势：手写一个工业级环形缓冲区

一句话结论：中断里只做「搬字节进缓冲区」，主循环里再做解析——两者之间的桥梁就是一个环形缓冲区（Ring Buffer）。

一、完整工程下载

压缩包内含全部源码、platformio.ini、Makefile、README.md，解压即用，不需要额外配置。

下载 ring-buffer.zip　（8.2 KB，共 7 个文件）

.gitignore
Makefile
README.md
include/
  ring_buffer.h
platformio.ini
src/
  ring_buffer.c
test/
  test_ring_buffer.c

为什么一定要用环形缓冲区

新手最常写出来的串口接收代码长这样：

void USART1_IRQHandler(void) {
    uint8_t b = USART1->DR;
    /* 在这里直接解析协议、printf、甚至控制电机 */
}

这段代码有三个致命问题：

中断里耗时太长。解析一帧协议可能几十微秒，串口 115200 时每个字节间隔只有 87µs，

一旦被高优先级中断抢占，下一个字节就丢了。

主循环与中断共享数据没有保护，buf[i++] 这种写法在两边同时执行时会覆盖数据。
帧长不确定，固定数组要么溢出要么浪费。

正确做法是把两者解耦：

USART ISR  -->  [ 环形缓冲区 ]  -->  主循环解析任务
只做 push 一个字节             只做 pop，处理慢也没关系

只要 写入速度的平均值 不超过 读取速度的平均值，缓冲区不溢出，就永远不会丢数据。

环形缓冲区的三个关键设计点

1. 用「留一个空位」区分满和空

如果只用 head == tail 表示状态，那么「满」和「空」的条件完全一样，无法区分。经典解法有三种，这里用最省事的 留一格空位：

状态	判断条件	可写字节数
空	`head == tail`	容量
满	`(head + 1) % size == tail`	0
已用	`(head - tail + size) % size`	容量 - 1

代价是浪费 1 个字节，换来判断逻辑不用加锁。

2. 单生产者单消费者天然无锁

head 只被写入方（中断）修改，tail 只被读取方（主循环）修改。两边各自读一下对方的指针即可判断状态，不需要关中断。注意两个变量必须声明为 volatile，否则编译器会把它们缓存进寄存器，导致死循环。

3. 容量取 2 的幂，但 head/tail 不要取模

这是一个很容易踩的坑：网上很多实现把 head/tail 对容量取模（head = (head + 1) % size），这样虽然索引方便，但「已用字节数」就 head - tail 变成负数，只能再写一堆 if 判断回绕。

正确做法是分开两件事：

/* 索引 = 单调递增的序号 & (size-1)，这一步取模是为了定位数组下标 */
rb->mem[rb->head & (rb->size - 1u)] = byte;
/* head 本身保持单调递增（uint16_t 自然回绕），
 * 「已用字节数」直接用无符号差值，天然正确、无分支 */
rb->head = (uint16_t)(rb->head + 1u);

于是关键的三个判断变成一行：

count = (uint16_t)(head - tail);            /* 已用 */
is_full  = (count == size - 1);             /* 满：留一格空位 */
is_empty = (head == tail);                  /* 空 */

本工程在写第一版时就是在这里写错了（把 head/tail 也取了模），结果第 8 个字节写满后就判断不出「满」。主机端单元测试直接把它抓了出来—— 这就是为什么每个工程都应该有一份能在电脑上跑的测试。

这个工程里包含什么

ring_buffer.c/h：完整实现，支持单字节读写、批量读写、peek、discard、查询剩余空间
两种模式：丢弃新数据（默认，保护已有数据）和 覆盖最旧数据（最新数据优先）
test/test_ring_buffer.c：主机端单元测试，覆盖回绕、溢出、批量读写、边界条件
platformio.ini：STM32F103C8T6（Blue Pill）工程配置

怎么用在你自己的项目里

static ring_buffer_t rx;            /* 全局，中断和主循环共用 */
static uint8_t rx_mem[256];

void uart_init(void) {
    ring_buffer_init(&rx, rx_mem, sizeof(rx_mem));
    /* ... 配置 USART，开 RXNE 中断 ... */
}

void USART1_IRQHandler(void) {
    if (USART1->SR & USART_SR_RXNE) {
        ring_buffer_put(&rx, (uint8_t)(USART1->DR & 0xFF));
    }
}

int main(void) {
    uart_init();
    for (;;) {
        uint8_t buf[32];
        size_t n = ring_buffer_pop(&rx, buf, sizeof(buf));
        for (size_t i = 0; i < n; i++) {
            protocol_feed(buf[i]);   /* 解析状态机，慢慢来 */
        }
    }
}

调试要点

缓冲区开多大？ 按「最坏情况下两次主循环之间的字节数」的 2~3 倍来算。

115200 波特率约 11.5KB/s，主循环最坏卡 5ms，就需要 ≥ 64 字节。

怎么发现溢出？ 实现里维护一个 overflow_cnt，主循环里定期打印。

只要它不为 0，说明主循环太慢或者中断太频繁。

不要在中断里调用 pop，那会把无锁设计破坏掉（两个消费者了）。

进阶方向

加上 DMA + 空闲中断（IDLE），让硬件帮你把一整帧搬进缓冲区，CPU 几乎零开销
换成「块状」环形缓冲区（每块带长度头），更适合变长帧
在多任务环境（FreeRTOS）下加入 taskENTER_CRITICAL() 保护，变成多生产者多消费者模型

完整代码

`Makefile`

CC      ?= gcc
CFLAGS  ?= -std=c99 -Wall -Wextra -O2 -Iinclude
LDLIBS  ?= 
SRC      = src/ring_buffer.c
TEST     = test/test_ring_buffer.c

ifeq ($(OS),Windows_NT)
EXT = .exe
endif
BIN = build/test$(EXT)

all: run

$(BIN): $(SRC) $(TEST)
	@mkdir -p build
	$(CC) $(CFLAGS) $(SRC) $(TEST) -o $(BIN) $(LDLIBS)

run: $(BIN)
	@$(BIN)

clean:
	rm -rf build

.PHONY: all run clean

`include/ring_buffer.h`

/**
 * ring_buffer.h - 单生产者/单消费者 无锁环形缓冲区
 *
 * 使用约定（务必遵守，否则无锁前提被破坏）：
 *   生产者（通常是串口中断）只调用 ring_buffer_put / ring_buffer_push
 *   消费者（通常是主循环）只调用 ring_buffer_pop / ring_buffer_peek / ring_buffer_discard
 *   双方都可以调用 ring_buffer_count / ring_buffer_free / ring_buffer_is_empty
 */
#ifndef RING_BUFFER_H
#define RING_BUFFER_H

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

#ifdef __cplusplus
extern "C" {
#endif

/* 容量必须是 2 的幂，便于用位与代替取模 */
typedef enum {
    RB_MODE_DROP_NEW = 0, /* 满时丢弃新数据（保护已有数据，推荐用于协议帧） */
    RB_MODE_OVERWRITE = 1 /* 满时覆盖最旧数据（永远拿到最新数据，推荐用于传感器流） */
} ring_buffer_mode_t;

typedef struct {
    uint8_t *mem;         /* 用户提供的存储区 */
    uint16_t size;        /* 容量，必须为 2 的幂 */
    volatile uint16_t head; /* 单调递增的写序号（uint16 自然回绕），仅生产者修改 */
    volatile uint16_t tail; /* 单调递增的读序号，仅消费者修改 */
    ring_buffer_mode_t mode;
    volatile uint32_t overflow_cnt; /* 因满而丢弃/覆盖的次数 */
    uint32_t total_written;
    uint32_t total_read;
} ring_buffer_t;

/**
 * @brief 初始化。mem 由调用者提供（避免动态分配）。
 * @return true 成功；false 表示 size 不是 2 的幂或为 0
 */
bool ring_buffer_init(ring_buffer_t *rb, uint8_t *mem, uint16_t size,
                      ring_buffer_mode_t mode);

void ring_buffer_reset(ring_buffer_t *rb);

/* 生产者侧 */
bool ring_buffer_put(ring_buffer_t *rb, uint8_t byte);
uint16_t ring_buffer_push(ring_buffer_t *rb, const uint8_t *data, uint16_t len);

/* 消费者侧 */
bool ring_buffer_get(ring_buffer_t *rb, uint8_t *out);
uint16_t ring_buffer_pop(ring_buffer_t *rb, uint8_t *out, uint16_t max_len);
bool ring_buffer_peek(const ring_buffer_t *rb, uint8_t *out); /* 不取出，只看 */
void ring_buffer_discard(ring_buffer_t *rb, uint16_t len);    /* 丢弃 len 字节 */

/* 通用查询 */
uint16_t ring_buffer_count(const ring_buffer_t *rb); /* 已存字节数 */
uint16_t ring_buffer_free(const ring_buffer_t *rb);  /* 还能写多少（留一格空位） */
bool ring_buffer_is_empty(const ring_buffer_t *rb);
bool ring_buffer_is_full(const ring_buffer_t *rb);

#ifdef __cplusplus
}
#endif

#endif /* RING_BUFFER_H */

`platformio.ini`

[platformio]
default_envs = bluepill

[env:bluepill]
platform = ststm32
board = bluepill_f103c8
framework = arduino
upload_protocol = stlink
monitor_speed = 115200
build_flags =
    -Wall
    -Wextra
    -Isrc
lib_ldf_mode = deep+

`src/ring_buffer.c`

#include "ring_buffer.h"

static bool is_power_of_two(uint16_t v)
{
    return v != 0u && (uint16_t)(v & (uint16_t)(v - 1u)) == 0u;
}

bool ring_buffer_init(ring_buffer_t *rb, uint8_t *mem, uint16_t size,
                      ring_buffer_mode_t mode)
{
    if (rb == NULL || mem == NULL || !is_power_of_two(size)) {
        return false;
    }
    rb->mem = mem;
    rb->size = size;
    rb->head = 0;
    rb->tail = 0;
    rb->mode = mode;
    rb->overflow_cnt = 0;
    rb->total_written = 0;
    rb->total_read = 0;
    return true;
}

void ring_buffer_reset(ring_buffer_t *rb)
{
    if (rb == NULL) {
        return;
    }
    rb->head = 0;
    rb->tail = 0;
}

/* head 与 tail 共同构成自然回绕的 16 位序列号，用相对差值做所有判断，
 * 这样即使 head 回绕多次也依然正确。 */
uint16_t ring_buffer_count(const ring_buffer_t *rb)
{
    return (uint16_t)(rb->head - rb->tail);
}

uint16_t ring_buffer_free(const ring_buffer_t *rb)
{
    return (uint16_t)((rb->size - 1u) - ring_buffer_count(rb));
}

bool ring_buffer_is_empty(const ring_buffer_t *rb)
{
    return rb->head == rb->tail;
}

bool ring_buffer_is_full(const ring_buffer_t *rb)
{
    return ring_buffer_count(rb) == (uint16_t)(rb->size - 1u);
}

bool ring_buffer_put(ring_buffer_t *rb, uint8_t byte)
{
    if (ring_buffer_is_full(rb)) {
        rb->overflow_cnt++;
        if (rb->mode == RB_MODE_DROP_NEW) {
            return false;
        }
        /* 覆盖模式：把最旧的扔掉 */
        rb->tail = (uint16_t)(rb->tail + 1u);
    }
    rb->mem[rb->head & (uint16_t)(rb->size - 1u)] = byte;
    rb->head = (uint16_t)(rb->head + 1u);
    rb->total_written++;
    return true;
}

uint16_t ring_buffer_push(ring_buffer_t *rb, const uint8_t *data, uint16_t len)
{
    uint16_t n = 0;
    if (data == NULL) {
        return 0;
    }
    while (n < len && ring_buffer_put(rb, data[n])) {
        n++;
    }
    /* 覆盖模式下 ring_buffer_put 永远返回 true，需要额外判断 */
    if (rb->mode == RB_MODE_OVERWRITE) {
        n = len;
    }
    return n;
}

uint16_t ring_buffer_pop(ring_buffer_t *rb, uint8_t *out, uint16_t max_len)
{
    uint16_t n = 0;
    if (out == NULL) {
        return 0;
    }
    while (n < max_len && ring_buffer_get(rb, &out[n])) {
        n++;
    }
    return n;
}

bool ring_buffer_get(ring_buffer_t *rb, uint8_t *out)
{
    if (ring_buffer_is_empty(rb)) {
        return false;
    }
    if (out != NULL) {
        *out = rb->mem[rb->tail & (uint16_t)(rb->size - 1u)];
    }
    rb->tail = (uint16_t)(rb->tail + 1u);
    rb->total_read++;
    return true;
}

bool ring_buffer_peek(const ring_buffer_t *rb, uint8_t *out)
{
    if (ring_buffer_is_empty(rb) || out == NULL) {
        return false;
    }
    *out = rb->mem[rb->tail & (uint16_t)(rb->size - 1u)];
    return true;
}

void ring_buffer_discard(ring_buffer_t *rb, uint16_t len)
{
    while (len-- > 0u && !ring_buffer_is_empty(rb)) {
        rb->tail = (uint16_t)(rb->tail + 1u);
        rb->total_read++;
    }
}

`test/test_ring_buffer.c`

/**
 * 主机端单元测试：gcc -std=c99 -Wall -Wextra -Iinclude src/ring_buffer.c test/test_ring_buffer.c -o build/test
 */
#include <stdio.h>
#include <string.h>

#include "ring_buffer.h"

static int g_pass = 0;
static int g_fail = 0;

#define CHECK(cond, msg)                                        \
    do {                                                        \
        if (cond) {                                             \
            g_pass++;                                           \
        } else {                                                \
            g_fail++;                                           \
            printf("  [FAIL] %s  (line %d)\n", msg, __LINE__);  \
        }                                                       \
    } while (0)

static uint8_t mem[8];

static void test_init(void)
{
    ring_buffer_t rb;
    printf("[1] 初始化校验\n");
    CHECK(ring_buffer_init(&rb, mem, 8, RB_MODE_DROP_NEW), "容量 8 应成功");
    CHECK(!ring_buffer_init(&rb, mem, 9, RB_MODE_DROP_NEW), "容量 9 应失败");
    CHECK(!ring_buffer_init(&rb, mem, 0, RB_MODE_DROP_NEW), "容量 0 应失败");
    CHECK(ring_buffer_is_empty(&rb), "初始化后应为空");
    CHECK(ring_buffer_count(&rb) == 0, "count 应为 0");
    CHECK(ring_buffer_free(&rb) == 7, "可用空间应为 size-1 = 7");
}

static void test_basic(void)
{
    ring_buffer_t rb;
    uint8_t v = 0;
    ring_buffer_init(&rb, mem, 8, RB_MODE_DROP_NEW);

    printf("[2] 基本读写\n");
    CHECK(ring_buffer_put(&rb, 'A'), "写入 A");
    CHECK(ring_buffer_put(&rb, 'B'), "写入 B");
    CHECK(ring_buffer_count(&rb) == 2, "已存 2 字节");
    CHECK(ring_buffer_get(&rb, &v) && v == 'A', "读出 A（FIFO）");
    CHECK(ring_buffer_get(&rb, &v) && v == 'B', "读出 B（FIFO）");
    CHECK(!ring_buffer_get(&rb, &v), "空了应返回 false");
    CHECK(ring_buffer_is_empty(&rb), "应为空");
}

static void test_wraparound(void)
{
    ring_buffer_t rb;
    uint8_t v = 0;
    ring_buffer_init(&rb, mem, 8, RB_MODE_DROP_NEW);

    printf("[3] 回绕（连续读写 1000 轮，容量仅 7）\n");
    for (int round = 0; round < 1000; round++) {
        for (int i = 0; i < 7; i++) {
            CHECK(ring_buffer_put(&rb, (uint8_t)(round * 7 + i)), "填满");
        }
        CHECK(ring_buffer_is_full(&rb), "应为满");
        CHECK(!ring_buffer_put(&rb, 0xEE), "满时丢弃新数据");
        for (int i = 0; i < 7; i++) {
            CHECK(ring_buffer_get(&rb, &v), "取出");
            CHECK(v == (uint8_t)(round * 7 + i), "数据顺序正确");
        }
    }
    CHECK(rb.overflow_cnt == 1000, "溢出计数应为 1000");
    CHECK(rb.total_written == 7000, "累计写入应为 7000");
}

static void test_overwrite_mode(void)
{
    ring_buffer_t rb;
    uint8_t v = 0;
    ring_buffer_init(&rb, mem, 8, RB_MODE_OVERWRITE);

    printf("[4] 覆盖模式：永远保留最新数据\n");
    for (int i = 0; i < 10; i++) {
        ring_buffer_put(&rb, (uint8_t)('0' + i));
    }
    CHECK(rb.overflow_cnt == 3, "溢出 3 次");
    CHECK(ring_buffer_get(&rb, &v) && v == '3', "最旧的 '0'~'2' 已被覆盖");
    for (int i = 4; i < 10; i++) {
        CHECK(ring_buffer_get(&rb, &v) && v == (uint8_t)('0' + i), "顺序正确");
    }
}

static void test_bulk(void)
{
    ring_buffer_t rb;
    uint8_t out[16];
    const uint8_t in[] = {1, 2, 3, 4, 5, 6};
    ring_buffer_init(&rb, mem, 8, RB_MODE_DROP_NEW);

    printf("[5] 批量读写\n");
    CHECK(ring_buffer_push(&rb, in, 6) == 6, "批量写入 6 字节");
    CHECK(ring_buffer_count(&rb) == 6, "count = 6");
    CHECK(ring_buffer_push(&rb, in, 6) == 1, "只剩 1 格可写");
    CHECK(ring_buffer_pop(&rb, out, sizeof(out)) == 7, "批量取出 7 字节");
    CHECK(memcmp(out, in, 6) == 0 && out[6] == 1, "内容正确");
    CHECK(ring_buffer_is_empty(&rb), "已取空");

    CHECK(ring_buffer_push(&rb, in, 6) == 6, "再写 6");
    ring_buffer_discard(&rb, 3);
    CHECK(ring_buffer_count(&rb) == 3, "丢弃 3 后剩 3");
    CHECK(ring_buffer_get(&rb, &out[0]) && out[0] == 4, "剩余数据从 4 开始");
}

static void test_peek(void)
{
    ring_buffer_t rb;
    uint8_t v = 0;
    ring_buffer_init(&rb, mem, 8, RB_MODE_DROP_NEW);
    printf("[6] peek 不消耗数据\n");
    ring_buffer_put(&rb, 0x5A);
    CHECK(ring_buffer_peek(&rb, &v) && v == 0x5A, "peek 得到 0x5A");
    CHECK(ring_buffer_count(&rb) == 1, "count 仍为 1");
    CHECK(ring_buffer_get(&rb, &v) && v == 0x5A, "get 得到 0x5A");
}

static void test_interleaved(void)
{
    ring_buffer_t rb;
    uint8_t v = 0;
    uint32_t written = 0;
    uint32_t read = 0;
    ring_buffer_init(&rb, mem, 16, RB_MODE_DROP_NEW);

    printf("[7] 生产者/消费者交错（模拟中断与主循环）\n");
    for (int i = 0; i < 500; i++) {
        if (!ring_buffer_is_full(&rb)) {
            ring_buffer_put(&rb, (uint8_t)(written & 0xFF));
            written++;
        }
        if ((i % 3) == 0 && ring_buffer_get(&rb, &v)) {
            CHECK(v == (uint8_t)(read & 0xFF), "交错下数据不乱序");
            read++;
        }
    }
    CHECK(ring_buffer_count(&rb) == written - read, "最终 count 一致");
    printf("     写入 %u 字节，读出 %u 字节，溢出 %u 次\n",
           rb.total_written, rb.total_read, rb.overflow_cnt);
}

int main(void)
{
    printf("===== ring_buffer 主机端测试 =====\n");
    test_init();
    test_basic();
    test_wraparound();
    test_overwrite_mode();
    test_bulk();
    test_peek();
    test_interleaved();
    printf("----- 通过 %d 项，失败 %d 项 -----\n", g_pass, g_fail);
    return g_fail == 0 ? 0 : 1;
}

实测输出

下面这段输出是把上面的核心算法用 本机 gcc 真编译、真运行得到的（不含任何硬件依赖）：

===== ring_buffer 主机端测试 =====
[1] 初始化校验
[2] 基本读写
[3] 回绕（连续读写 1000 轮，容量仅 7）
[4] 覆盖模式：永远保留最新数据
[5] 批量读写
[6] peek 不消耗数据
[7] 生产者/消费者交错（模拟中断与主循环）
     写入 182 字节，读出 167 字节，溢出 0 次
----- 通过 23203 项，失败 0 项 -----

菜单

分享

嵌入式串口收发的正确姿势：手写一个工业级环形缓冲区

一、完整工程下载

为什么一定要用环形缓冲区

环形缓冲区的三个关键设计点

1. 用「留一个空位」区分满和空

2. 单生产者单消费者天然无锁

3. 容量取 2 的幂，但 head/tail 不要取模

这个工程里包含什么

怎么用在你自己的项目里

调试要点

进阶方向

完整代码

`Makefile`

`include/ring_buffer.h`

`platformio.ini`

`src/ring_buffer.c`

`test/test_ring_buffer.c`

实测输出

评论

端到端测试

温度传感器不准？B 参数公式、Steinhart-Hart 与多点标定的误差实测

步进电机为什么会“丢步”：梯形与 S 曲线加减速的完整实现

裸机也该有“操作系统”：80 行写一个软定时器任务调度器

给单片机写个能用的命令行：行编辑、历史、Tab 补全、ANSI 转义全都要

PID 从公式到能跑：给直流电机写一个带抗积分饱和的速度环

手写 Modbus RTU 从站：03/06/16 功能码 + 异常响应，用测试用例逐字节验证

固定块内存池：为什么嵌入式该用它替掉 malloc（附真实碎片实测）

现场升级不翻车：手写一个带 A/B 双分区与自动回滚的串口 IAP

软件 I2C 实战：开漏总线、时钟延展、扫描与 9 时钟解锁