第5章消息队列高级特性

课程目录第5章消息队列高级特性

5. 消息队列高级特性：优先级队列、超时机制、广播消息、批量消息处理

各位同学，咱们今天聊点硬核的。消息队列这东西，说白了就是个「中间人」。但光会收发消息，那只是入门。真正让系统变得灵活、高效、可靠的，是那些高级特性。我这些年做嵌入式项目，从智能家居到工业控制器，几乎每个系统都离不开这些技巧。

嗯，咱们一个一个来看。

5.1 优先级队列：让重要消息插队

你想想看，一个系统里同时来了好几条消息。有紧急的报警信号，也有普通的日志上报。如果一视同仁，先来后到，那报警信号可能就被堵在路上了。这怎么行？

优先级队列就是干这个的。它允许每条消息带一个「优先级标签」。高优先级的消息，可以插队到队列前面去。

核心思路： 用多个普通队列，每个队列对应一个优先级。发送消息时，根据优先级放入对应队列。接收时，总是先检查高优先级队列。

我在项目中遇到过这种情况：一个电机控制系统，过流保护信号必须毫秒级响应。而温度采集数据，慢个几百毫秒无所谓。我当时就用了三级优先级队列：紧急、高、普通。效果立竿见影。

代码实现上，我习惯用数组加链表的方式：

// 优先级队列结构
#define PRIORITY_LEVELS 3

typedef struct {
    MsgNode *head[PRIORITY_LEVELS];
    MsgNode *tail[PRIORITY_LEVELS];
} PriorityQueue;

// 发送：根据优先级入队
void prio_queue_send(PriorityQueue *q, Msg *msg, int priority) {
    // 插入到对应优先级的链表尾部
}

// 接收：从最高优先级开始检查
Msg* prio_queue_recv(PriorityQueue *q) {
    for (int i = PRIORITY_LEVELS - 1; i >= 0; i--) {
        if (q->head[i] != NULL) {
            // 从该优先级队列头部取出
            return dequeue(q, i);
        }
    }
    return NULL; // 所有队列都空
}

我的小技巧： 优先级不要设太多，3-5级就够用了。太多级反而增加调度开销，得不偿失。

5.2 超时机制：别让任务死等

消息队列有个常见问题：接收方如果一直等不到消息，怎么办？

死等？那系统就卡死了。我见过一个同事写的代码，接收消息时用了无限阻塞。结果某个模块挂了，整个系统都跟着瘫痪。嗯，这就是典型的「一颗老鼠屎坏了一锅粥」。

超时机制就是解决方案。给接收操作加一个时间限制。时间到了还没消息，就返回一个超时标志。调用方可以去做别的事，或者重试。

// 带超时的消息接收
typedef enum {
    MSG_OK,
    MSG_TIMEOUT,
    MSG_ERROR
} MsgResult;

MsgResult recv_with_timeout(MsgQueue *q, Msg *out, uint32_t timeout_ms) {
    uint32_t start = get_tick_ms();
    
    while (1) {
        if (queue_is_empty(q)) {
            // 检查是否超时
            if (get_tick_ms() - start >= timeout_ms) {
                return MSG_TIMEOUT;
            }
            // 让出CPU，避免忙等
            task_yield();
            continue;
        }
        
        *out = dequeue(q);
        return MSG_OK;
    }
}

我曾经踩过的坑： 超时时间设得太短，导致正常消息被误判为超时。后来我总结了一个原则：超时时间至少是消息处理时间的3倍。另外，记得在循环里加一点延时或任务切换，别让CPU空转。

5.3 广播消息：一对多，效率高

有时候，一条消息需要通知多个接收者。比如系统掉电预警，所有模块都得知道。如果逐个发送，代码冗余不说，还容易漏掉。

广播消息就是干这个的。一条消息发出去，所有订阅了这个消息类型的接收者都能收到。

实现方式其实不复杂。我常用的方法是「订阅者列表」：

// 广播消息结构
#define MAX_SUBSCRIBERS 16

typedef struct {
    MsgQueue *subscribers[MAX_SUBSCRIBERS];
    int count;
} BroadcastChannel;

// 订阅
void broadcast_subscribe(BroadcastChannel *ch, MsgQueue *q) {
    if (ch->count < MAX_SUBSCRIBERS) {
        ch->subscribers[ch->count++] = q;
    }
}

// 广播
void broadcast_send(BroadcastChannel *ch, Msg *msg) {
    for (int i = 0; i < ch->count; i++) {
        // 每个订阅者都复制一份消息
        Msg *copy = msg_copy(msg);
        queue_send(ch->subscribers[i], copy);
    }
}

注意： 广播时每条消息都要复制一份，否则多个接收者会共享同一块内存，容易出问题。我习惯用引用计数来管理消息的生命周期，避免内存泄漏。

5.4 批量消息处理：攒一波，一起干

有些场景下，消息来得特别快。比如传感器数据采集，每秒几百条。如果来一条处理一条，系统开销很大。上下文切换、函数调用、锁操作……这些都会拖慢速度。

批量处理就是攒够一批消息，一次性处理。这样能大幅提升吞吐量。

我做过一个数据记录仪项目，SD卡写入速度是瓶颈。单条写入太慢，我就用了批量消息：攒够64条数据，一次性写入SD卡。速度提升了将近10倍。

// 批量接收
int recv_batch(MsgQueue *q, Msg *buffer, int max_count, uint32_t timeout_ms) {
    int count = 0;
    uint32_t start = get_tick_ms();
    
    while (count < max_count) {
        MsgResult res = recv_with_timeout(q, &buffer[count], timeout_ms);
        if (res == MSG_OK) {
            count++;
        } else if (res == MSG_TIMEOUT) {
            // 超时了，有多少收多少
            break;
        }
    }
    
    return count; // 返回实际收到的数量
}

批量处理的关键： 找到合适的「批量大小」。太小了没效果，太大了延迟太高。我一般根据系统负载和内存情况，动态调整批量大小。比如系统空闲时批量小一点，忙时批量大一点。

5.5 知识体系总览

说了这么多，咱们用一张图来梳理一下。这张图展示了消息队列高级特性的核心逻辑和它们之间的关系。

这张图把四个高级特性串起来了。你看，它们不是孤立的。实际项目中，我经常把优先级队列和超时机制结合使用。高优先级消息的超时时间设短一点，低优先级的设长一点。这样既保证了紧急消息的及时性，又不会让系统因为等待低优先级消息而卡住。

好了，这一章的内容就到这里。优先级队列、超时机制、广播消息、批量处理，这四个特性用好，你的消息队列系统就能上一个台阶。下次咱们聊聊消息队列在实际项目中的架构设计。

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

📚 课程目录

5. 消息队列高级特性：优先级队列、超时机制、广播消息、批量消息处理

5.1 优先级队列：让重要消息插队

5.2 超时机制：别让任务死等

5.3 广播消息：一对多，效率高

5.4 批量消息处理：攒一波，一起干

5.5 知识体系总览