4、消息队列原理：生产者-消费者模型、队列数据结构（环形缓冲区、链表队列）、阻塞与非阻塞模式、消息优先级

消息队列这东西，说白了就是让不同任务之间能「递纸条」。我做了这么多年嵌入式，发现很多系统崩溃、数据错乱，根源都是任务之间没协调好。消息队列就是解决这个问题的经典方案。

4.1 生产者-消费者模型：解耦的核心思想

先聊聊这个模型。生产者负责产生数据，消费者负责处理数据。两者通过一个共享的队列来交互，彼此不需要知道对方的存在。

举个例子，你想想看：一个传感器采集任务（生产者）不断把数据塞进队列，一个显示任务（消费者）从队列里取数据刷新屏幕。这两个任务跑在不同的优先级上，甚至可能跑在不同的核上。如果没有队列，它们就得用全局变量加锁来同步，那代码写起来就痛苦了。

我在项目中遇到过一个问题：一个音频采集任务和编码任务直接通过全局缓冲区交互，结果采集速度偶尔比编码快，缓冲区被覆盖，声音就出现了爆音。后来改成消息队列，生产者只管往队列里放，消费者只管取，问题就解决了。

4.2 队列数据结构：环形缓冲区 vs 链表队列

队列的底层实现，我常用的就两种：环形缓冲区和链表队列。各有各的适用场景，别选错了。

4.2.1 环形缓冲区（Ring Buffer）

环形缓冲区，说白了就是一个固定大小的数组，加上两个指针：读指针和写指针。数据写满尾部后，自动绕回头部继续写。这种结构在嵌入式里用得特别多，因为它没有动态内存分配，性能稳定。

我习惯用环形缓冲区来处理高频数据流，比如串口接收、音频采样。为什么？因为它的内存是预分配的，不会出现内存碎片，而且入队出队操作都是 O(1) 的复杂度。

// 环形缓冲区简单实现
typedef struct {
    uint8_t *buffer;
    uint32_t size;
    volatile uint32_t head;  // 写指针
    volatile uint32_t tail;  // 读指针
} ring_buffer_t;

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

// 出队
int ring_buffer_pop(ring_buffer_t *rb, uint8_t *data) {
    if (rb->head == rb->tail) {
        return -1;  // 缓冲区空
    }
    *data = rb->buffer[rb->tail];
    rb->tail = (rb->tail + 1) % rb->size;
    return 0;
}

4.2.2 链表队列

链表队列就不一样了。它的节点是动态分配的，理论上可以无限增长。适合消息大小不固定、或者队列长度变化很大的场景。

我曾经在一个网络协议栈里用链表队列来管理待发送的数据包。因为数据包大小从几十字节到几千字节不等，用环形缓冲区很难分配合理的内存。链表队列就灵活多了，每个节点只存一个数据包，用完了就释放。

但要注意，链表队列有代价：每个节点需要额外的指针开销（通常8字节或16字节），而且频繁的 malloc/free 会导致内存碎片。在资源紧张的 MCU 上，我一般不用链表队列。

特性	环形缓冲区	链表队列
内存分配	静态预分配	动态分配
性能	O(1)，稳定	O(1)，但 malloc 耗时
内存碎片	无	有
适用场景	高频、固定大小数据	低频、大小不固定数据

4.3 阻塞与非阻塞模式

这个点很关键。阻塞和非阻塞，决定了任务在队列空或满时的行为。

阻塞模式：当队列为空时，消费者任务会被挂起，直到有数据入队才被唤醒。同样，队列满时生产者也会被挂起。这种模式适合任务之间有明确依赖关系的场景。

我记得在一个工业控制项目中，一个任务负责接收传感器数据，另一个任务负责计算控制量。如果计算任务比采集任务快，它就得等数据。用阻塞模式，计算任务在队列空时自动休眠，不浪费 CPU。

非阻塞模式：不管队列空还是满，操作立即返回。空时返回空标志，满时返回满标志。调用方需要自己处理这种情况。

非阻塞模式适合那些不能等待的场景，比如中断服务函数里。你想想看，中断里如果调用阻塞式出队，万一队列空了，整个系统就卡死了。

4.4 消息优先级

有些场景下，消息是有轻重缓急的。比如一个紧急报警消息，必须比普通日志消息先处理。这时候就需要消息优先级机制。

实现方式有两种：

• 多队列方案：为每个优先级分配一个独立的队列。发送时根据优先级选择队列，接收时从高优先级队列开始检查。这种方案实现简单，但需要管理多个队列。
• 单队列排序方案：所有消息放在一个队列里，但按优先级排序。每次插入时找到合适的位置，优先级高的排在前面。这种方案实现复杂，但节省内存。

我个人习惯用多队列方案。为什么？因为它的入队操作是 O(1)，而出队操作只需要遍历优先级列表，通常优先级数量很少（比如4级或8级），性能完全够用。

// 多优先级队列示例
#define PRIORITY_LEVELS 4

typedef struct {
    ring_buffer_t queues[PRIORITY_LEVELS];
} priority_queue_t;

// 发送消息
int priority_send(priority_queue_t *pq, uint8_t priority, uint8_t data) {
    if (priority >= PRIORITY_LEVELS) return -1;
    return ring_buffer_push(&pq->queues[priority], data);
}

// 接收消息（从最高优先级开始检查）
int priority_recv(priority_queue_t *pq, uint8_t *data) {
    for (int i = PRIORITY_LEVELS - 1; i >= 0; i--) {
        if (ring_buffer_pop(&pq->queues[i], data) == 0) {
            return i;  // 返回优先级
        }
    }
    return -1;  // 所有队列都空
}

嗯，消息队列的原理就这些。实际项目中，你可能会遇到各种变体，比如带超时的阻塞、多生产者多消费者、消息拷贝策略等等。但核心思想不变：通过一个中间队列，让生产者和消费者各司其职，互不干扰。

---

公众号:蓝海资料掘金营，微信deep3321