3. 事件队列设计：环形缓冲区实现，队列深度与溢出处理，多生产者单消费者模型

事件队列是整个事件驱动架构的心脏。说白了，所有任务调度都靠它来传递消息。如果队列设计得不好，系统跑起来就会各种诡异——丢事件、死锁、优先级反转，我都遇到过。

我个人习惯把事件队列比作一个环形跑道。生产者往跑道上放包裹，消费者从跑道上取包裹。跑道的长度是固定的，跑完了就从头再来。这就是环形缓冲区（Ring Buffer）的核心思想。

3.1 环形缓冲区的基本实现

环形缓冲区为什么受欢迎？因为它没有动态内存分配，没有链表节点开销，只有一块连续内存和两个指针。在嵌入式系统里，这简直就是黄金方案。

/* 环形缓冲区结构体 */
typedef struct {
    event_t *buffer;      /* 事件缓冲区指针 */
    uint32_t head;        /* 生产者写入位置 */
    uint32_t tail;        /* 消费者读取位置 */
    uint32_t size;        /* 缓冲区容量（必须是2的幂） */
    uint32_t count;       /* 当前事件数量 */
} ring_buffer_t;

/* 初始化 */
void rb_init(ring_buffer_t *rb, event_t *buf, uint32_t size) {
    rb->buffer = buf;
    rb->head = 0;
    rb->tail = 0;
    rb->size = size;
    rb->count = 0;
}

/* 写入事件 */
bool rb_put(ring_buffer_t *rb, event_t *evt) {
    if (rb->count >= rb->size) {
        return false;  /* 队列满了 */
    }
    rb->buffer[rb->head] = *evt;
    rb->head = (rb->head + 1) & (rb->size - 1);
    rb->count++;
    return true;
}

/* 读取事件 */
bool rb_get(ring_buffer_t *rb, event_t *evt) {
    if (rb->count == 0) {
        return false;  /* 队列空了 */
    }
    *evt = rb->buffer[rb->tail];
    rb->tail = (rb->tail + 1) & (rb->size - 1);
    rb->count--;
    return true;
}

3.2 队列深度怎么定？

这是每个项目都会问的问题。队列深度设大了浪费内存，设小了丢事件。我踩过这个坑，而且不止一次。

队列深度取决于三个因素：

• 事件产生速率：峰值情况下，单位时间能产生多少个事件？
• 事件消费速率：最坏情况下，处理一个事件要多久？
• 容忍的丢包率：丢一个事件系统会崩溃吗？

我一般用这个公式估算：

队列深度 = (最大生产速率 - 最小消费速率) × 最坏突发持续时间

举个例子：一个传感器每秒产生 1000 个事件，CPU 每秒最多处理 800 个，突发持续 100ms。那么队列深度至少需要 (1000 - 800) × 0.1 = 20。但实际我会留 2 倍余量，设成 64。

场景	推荐队列深度	说明
按键输入	8 ~ 16	人按键盘的速度很慢，深度不用太大
传感器数据流	64 ~ 256	看采样率和处理时间，通常要留余量
网络数据包	128 ~ 512	突发流量大，深度不够会丢包
CAN 总线消息	32 ~ 128	取决于总线负载和节点数量

3.3 溢出处理策略

队列满了怎么办？有三种常见策略：

1. 丢弃新事件：最简单，但可能丢失关键信息。
2. 覆盖最旧事件：适合传感器数据，旧数据不如新数据重要。
3. 阻塞等待：生产者等消费者腾出空间，但中断里不能用。

我个人习惯用「丢弃新事件 + 错误计数」。这样既不会破坏系统状态，又能通过错误计数知道系统过载了。

/* 带溢出计数的写入 */
bool rb_put_safe(ring_buffer_t *rb, event_t *evt, uint32_t *overflow_cnt) {
    if (rb->count >= rb->size) {
        if (overflow_cnt) {
            (*overflow_cnt)++;
        }
        return false;  /* 丢弃新事件，但记录溢出次数 */
    }
    rb->buffer[rb->head] = *evt;
    rb->head = (rb->head + 1) & (rb->size - 1);
    rb->count++;
    return true;
}

3.4 多生产者单消费者模型

这是嵌入式系统里最常见的场景。多个中断源（生产者）往队列里塞事件，一个主循环（消费者）从队列里取事件处理。

这里有个关键问题：多生产者同时写入怎么办？

如果两个中断同时触发，同时修改 head 指针和 count 变量，数据就乱了。解决方案是保证写入操作的原子性。

对于单核 MCU，最简单的办法是关中断：

/* 多生产者安全写入 */
bool rb_put_from_isr(ring_buffer_t *rb, event_t *evt) {
    uint32_t saved = __disable_irq();  /* 关中断 */
    bool result = rb_put(rb, evt);
    __enable_irq();                    /* 恢复中断 */
    return result;
}

但关中断的时间要尽量短。我见过有人关中断 100 微秒，结果系统定时器都跑偏了。环形缓冲区的写入操作本来就快，关中断时间通常在 1 微秒以内，完全没问题。

对于多核 MCU 或者 RTOS 环境，可以用原子操作：

/* 使用原子操作 */
bool rb_put_atomic(ring_buffer_t *rb, event_t *evt) {
    uint32_t current_count;
    do {
        current_count = __atomic_load_n(&rb->count, __ATOMIC_SEQ_CST);
        if (current_count >= rb->size) {
            return false;  /* 满了 */
        }
    } while (!__atomic_compare_exchange_n(&rb->count, ¤t_count, 
                                          current_count + 1, 
                                          false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST));
    /* 这里可以安全写入，因为 count 已经被我们占用了 */
    rb->buffer[rb->head] = *evt;
    rb->head = (rb->head + 1) & (rb->size - 1);
    return true;
}

3.5 环形缓冲区结构图

下面这张图展示了环形缓冲区在多生产者单消费者模型下的工作流程：

3.6 调试技巧

队列出问题的时候，现象往往很隐蔽。事件偶尔丢失，系统偶尔卡顿。我分享几个调试经验：

• 加一个队列水位监控：每隔一段时间记录 count 的最大值。如果接近 size，说明队列深度不够。
• 溢出计数一定要暴露出来：我习惯把溢出计数放在调试信息里，随时能看。
• 用逻辑分析仪抓事件时间戳：看看生产者写入和消费者读取的时间差，能发现很多问题。