4. 事件队列的设计与实现

事件队列,说白了就是整个事件驱动架构的「血管」。没有它,事件就流不动。我见过不少团队,CPU选得挺强,任务划分也合理,结果队列设计太随意,系统跑起来就是各种丢事件、优先级反转。嗯,这一章咱们就把队列这件事彻底聊透。

4.1 环形缓冲区原理

先问一个问题:为什么嵌入式里几乎不用链表做队列?

原因很简单——嵌入式环境不允许动态内存分配。你想想看,一个malloc下去,万一碎片了、万一分配失败了,系统怎么办?所以,我们得用静态分配的环形缓冲区。

环形缓冲区的核心思想:一块固定大小的数组,两个指针(读指针和写指针),绕圈走。写指针追上读指针,说明满了;读指针追上写指针,说明空了。

关键判断条件:

  • 空:(head == tail)
  • 满:((tail + 1) % size == head)

注意这里「满」的判断会浪费一个元素空间,用来区分空和满。我个人习惯用这个方案,简单可靠。

// 环形缓冲区核心结构
typedef struct {
    uint8_t *buffer;      // 数据缓冲区
    uint32_t size;        // 缓冲区大小
    uint32_t head;        // 读指针
    uint32_t tail;        // 写指针
} ring_buffer_t;

// 写入一个字节
int ring_buffer_write(ring_buffer_t *rb, uint8_t data) {
    uint32_t next = (rb->tail + 1) % rb->size;
    if (next == rb->head) {
        return -1;  // 缓冲区满
    }
    rb->buffer[rb->tail] = data;
    rb->tail = next;
    return 0;
}

// 读取一个字节
int ring_buffer_read(ring_buffer_t *rb, uint8_t *data) {
    if (rb->head == rb->tail) {
        return -1;  // 缓冲区空
    }
    *data = rb->buffer[rb->head];
    rb->head = (rb->head + 1) % rb->size;
    return 0;
}

避坑指南:我曾经在一个电机控制项目里,直接用取模运算做环形缓冲区。结果发现取模在ARM Cortex-M0上慢得离谱。后来改成位与运算(要求size是2的幂),性能直接翻倍。

4.2 静态队列与动态队列

这里我直接给结论:嵌入式事件队列,99%的场景用静态队列就够了。

对比项 静态队列 动态队列
内存分配 编译时固定 运行时malloc/free
确定性 完全确定 不确定(碎片风险)
最大深度 固定上限 受堆大小限制
适用场景 绝大多数嵌入式系统 Linux用户态、原型验证
实时性 O(1) 无抖动 可能因GC或碎片抖动

你可能会问:那动态队列什么时候用?我在做Linux上的协议栈原型时用过,因为事件类型不确定、数量也不确定,用动态队列省事。但一旦进入产品阶段,全部换成静态。

4.3 队列深度计算

这是实际项目中最容易翻车的地方。队列深度设小了,丢事件;设大了,浪费RAM。怎么算?

我总结了一个经验公式:

队列深度 = (最大事件产生速率 × 最大处理延迟) / 事件大小 + 安全余量

举个例子:

  • 一个传感器每秒产生1000个事件
  • 每个事件8字节
  • 最坏情况下,处理一个事件需要5ms
  • 那么:1000 × 0.005 = 5个事件在队列中
  • 加上50%安全余量 → 8个事件
  • 每个事件8字节 → 64字节缓冲区

注意:我曾经在一个CAN总线项目中,按平均速率算队列深度,结果现场出现突发流量,队列直接爆了。从那以后,我坚持用「最坏情况」算,而不是平均值。安全余量至少留50%,关键系统留100%。

4.4 多生产者单消费者模型

这是嵌入式事件队列最经典的模型。多个中断服务程序(生产者)往队列里扔事件,一个主循环(消费者)从队列里取事件处理。

为什么是单消费者?因为多消费者需要锁,而嵌入式里用锁,搞不好就死锁。单消费者模型天然无锁,只要保证生产者的写操作是原子的就行。

// 多生产者单消费者 - 无锁实现
typedef struct {
    event_t events[QUEUE_SIZE];
    volatile uint32_t head;  // 消费者修改
    volatile uint32_t tail;  // 生产者修改
    uint32_t size;
} event_queue_t;

// 生产者(可在中断中调用)
int event_queue_post(event_queue_t *q, const event_t *evt) {
    uint32_t next = (q->tail + 1) % q->size;
    if (next == q->head) {
        return -1;  // 队列满
    }
    q->events[q->tail] = *evt;
    __DMB();  // 数据内存屏障,确保写完成
    q->tail = next;
    return 0;
}

// 消费者(在主循环中调用)
int event_queue_fetch(event_queue_t *q, event_t *evt) {
    if (q->head == q->tail) {
        return -1;  // 队列空
    }
    *evt = q->events[q->head];
    q->head = (q->head + 1) % q->size;
    return 0;
}

关键点:注意volatile关键字和内存屏障。我见过一个同事,代码逻辑完全正确,但就是偶尔丢事件。查了两天,发现是编译器优化把tail指针的写操作给重排了。加上__DMB()后问题解决。

4.5 整体架构图

下面这张图展示了事件队列在整个系统中的位置和流转关系:

事件队列架构图 事件队列架构图 中断服务程序1 中断服务程序2 中断服务程序3 多生产者 环形缓冲区 head → 读指针 tail → 写指针 size = 2^n 主循环 单消费者 关键设计要点: 1. 生产者之间不需要互斥,因为每个中断独立写tail指针 2. 消费者只读head,生产者只读tail,不会冲突 3. 必须使用volatile和内存屏障,防止编译器/CPU重排 4. 队列深度按最坏情况计算,安全余量留50%以上

4.6 实际项目中的避坑清单

  1. 队列深度别用平均值算——用最坏情况,否则现场必爆
  2. size必须是2的幂——取模运算用位与替代,性能差好几倍
  3. volatile不能省——编译器优化会搞乱你的指针
  4. 内存屏障要加对地方——写tail之前确保数据已经写入缓冲区
  5. 中断里别做复杂操作——只post事件,处理逻辑放主循环

血的教训:我曾经在一个工业控制器项目里,为了省几个字节的RAM,把队列深度压到刚好够用。结果现场出现一次异常中断风暴,队列瞬间填满,关键事件全部丢失,设备直接失控。从那以后,我的队列深度至少留一倍余量。

嗯,事件队列这部分就聊到这儿。记住一句话:队列是事件驱动架构的命脉,设计得保守一点,比设计得激进要安全得多。