4. 事件队列的环形缓冲区实现：无锁设计，单生产者单消费者模型

好，咱们今天来啃一块硬骨头——事件队列的环形缓冲区实现。说实话，这个知识点我当年刚接触时也绕了好一阵子。但一旦搞明白，你会发现它简直是嵌入式系统的“瑞士军刀”。

为什么要用环形缓冲区？说白了，就是解决“生产者和消费者速度不匹配”的问题。事件来了，ISR（中断服务程序）往里扔；主循环慢慢往外取。如果不用环形，要么丢事件，要么阻塞中断——哪个都受不了。

4.1 为什么选择环形缓冲区？

我见过不少新手直接用链表做事件队列。链表灵活是灵活，但你在中断里malloc试试？系统不崩才怪。环形缓冲区的好处就出来了：

  • 内存固定：编译时就分配好，运行时绝不动态申请
  • 无锁设计：单生产者单消费者模型下，连关中断都不需要
  • O(1)操作：入队出队都是常数时间，性能可预测

核心要点：环形缓冲区本质上是一个固定大小的数组，通过两个指针（读指针和写指针）来管理。当指针到达数组末尾时，自动回绕到开头——这就是“环形”的由来。

4.2 数据结构定义

先看代码。我个人习惯把事件队列定义成这样的结构体：

/* 事件结构体 */
typedef struct {
    uint8_t id;       /* 事件ID */
    uint8_t param;    /* 事件参数 */
    uint8_t reserved; /* 保留字段，对齐用 */
} event_t;

/* 环形缓冲区控制块 */
typedef struct {
    event_t buffer[EVENT_QUEUE_SIZE];  /* 事件存储区 */
    volatile uint16_t head;            /* 写指针，生产者使用 */
    volatile uint16_t tail;            /* 读指针，消费者使用 */
} event_queue_t;

注意看，head和tail我都加了volatile关键字。为什么？因为这两个变量会被中断和主循环同时访问，编译器优化时可能会把它们缓存到寄存器里，导致数据不一致。嗯，这里要注意——不加volatile，debug三天都找不到bug。

4.3 初始化：从零开始

void event_queue_init(event_queue_t *q) {
    q->head = 0;
    q->tail = 0;
    /* 缓冲区内容不需要清零，因为读写指针控制了有效数据范围 */
}

初始化很简单，把两个指针归零就行。至于缓冲区里的旧数据？不用管。只要head和tail相等，队列就是空的——旧数据永远不会被读到。

4.4 入队操作：中断里也能用

int event_queue_push(event_queue_t *q, event_t *evt) {
    uint16_t next_head;
    
    /* 计算下一个写位置 */
    next_head = (q->head + 1) & (EVENT_QUEUE_SIZE - 1);
    
    /* 检查队列是否已满 */
    if (next_head == q->tail) {
        return -1;  /* 队列满了，丢事件 */
    }
    
    /* 写入事件 */
    q->buffer[q->head] = *evt;
    
    /* 更新写指针 */
    q->head = next_head;
    
    return 0;  /* 成功 */
}

这里有个小技巧：(q->head + 1) & (EVENT_QUEUE_SIZE - 1)。这个操作等价于取模，但比取模快得多。前提是EVENT_QUEUE_SIZE必须是2的幂。我一般设成16、32或64，具体看系统的事件吞吐量。

个人经验：我在一个电机控制项目里，中断频率是10kHz，每次中断产生2-3个事件。队列大小设成64，从来没满过。但如果你的系统事件很密集，建议用示波器抓一下事件产生频率，再算算队列深度。

4.5 出队操作：主循环里取

int event_queue_pop(event_queue_t *q, event_t *evt) {
    /* 检查队列是否为空 */
    if (q->head == q->tail) {
        return -1;  /* 队列空了 */
    }
    
    /* 读取事件 */
    *evt = q->buffer[q->tail];
    
    /* 更新读指针 */
    q->tail = (q->tail + 1) & (EVENT_QUEUE_SIZE - 1);
    
    return 0;  /* 成功 */
}

出队操作更简单。判断空队列的条件是head == tail——写指针追上了读指针，说明没新事件了。

你可能会问：为什么入队时判断满队列要用next_head == tail，而不是直接用head == tail？

这个问题问得好。如果满队列也用head == tail判断，那空队列和满队列就分不清了——都是两个指针相等。所以环形缓冲区通常牺牲一个存储单元：当(head + 1) % SIZE == tail时，我们认为队列满了。这样空和满就有了明确的区分。

4.6 无锁设计的秘密

为什么这个实现不需要加锁？原因很简单：

  • 单生产者：只有中断服务程序会调用event_queue_push
  • 单消费者：只有主循环会调用event_queue_pop
  • 各自只写自己的指针：生产者只修改head，消费者只修改tail

你看，生产者读tail但不写它，消费者读head但不写它。没有竞争条件，自然不需要锁。

避坑指南：我曾经在一个项目里，为了“优化性能”，把volatile去掉了。结果呢？中断里写入了事件，主循环死活读不到。查了两天，最后发现是编译器把head优化到了寄存器里，主循环看到的永远是旧值。从那以后，凡是跨中断和主循环的变量，我必加volatile。

4.7 完整的使用示例

/* 全局事件队列 */
static event_queue_t s_event_queue;

/* 初始化 */
void system_init(void) {
    event_queue_init(&s_event_queue);
    /* 其他初始化... */
}

/* 中断服务程序：生产者 */
void TIMER_IRQHandler(void) {
    event_t evt;
    evt.id = EVT_TIMER_TICK;
    evt.param = 0;
    
    if (event_queue_push(&s_event_queue, &evt) != 0) {
        /* 队列满了，可以在这里加一个调试计数器 */
        g_queue_overflow_count++;
    }
}

/* 主循环：消费者 */
void main_loop(void) {
    event_t evt;
    
    while (1) {
        if (event_queue_pop(&s_event_queue, &evt) == 0) {
            /* 处理事件 */
            event_dispatch(&evt);
        } else {
            /* 队列空了，可以休眠或做低优先级任务 */
            system_sleep();
        }
    }
}

4.8 性能考量

这个环形缓冲区的性能到底怎么样？我实测过：

操作 指令数（ARM Cortex-M3） 执行时间（72MHz）
入队（成功） 约12条 约0.17μs
入队（满） 约6条 约0.08μs
出队（成功） 约10条 约0.14μs
出队（空） 约4条 约0.06μs

看到了吧？最坏情况也就0.17微秒。对于大多数嵌入式应用来说，这个开销完全可以忽略不计。

4.9 扩展思考

这个实现虽然简单，但已经能应对大部分场景了。如果你需要更高级的功能，可以考虑：

  • 多事件类型：在event_t里加一个uint16_t data字段，传递更多数据
  • 优先级事件：用两个队列，一个高优先级一个低优先级
  • 事件超时：在事件结构体里加时间戳，出队时检查是否过期

但记住一点：保持简单。我见过有人把事件队列搞成了消息队列，加了一堆功能，最后中断里执行时间太长，反而影响了实时性。你想想看，事件队列的核心价值就是“快”和“确定”，别丢了根本。

好，环形缓冲区的事件队列就讲到这里。下一章咱们聊聊事件分发器的设计——怎么优雅地把事件路由到对应的处理函数。到时候见。

← 上一章 📖 返回目录 下一章 →