4. 消息队列与邮箱：阻塞与超时、生产者-消费者模型、串口数据分发实战
消息队列和邮箱，是RTOS里最常用的任务间通信手段。说白了，它们就是给任务之间递纸条的“信箱”。但怎么递、递不过去怎么办、谁放谁取，这里头有不少门道。
我个人习惯把消息队列看作一个环形缓冲区加上任务等待列表。你往队列里发消息，如果队列满了，发送任务可以选择阻塞等待；你去队列里取消息，如果队列空了，接收任务也可以选择阻塞等待。这个“阻塞”和“超时”机制，就是RTOS事件驱动调度的精髓之一。
4.1 消息队列的核心机制
先看消息队列的数据结构长什么样。我简化一下核心字段：
typedef struct {
void *buffer; // 消息缓冲区
uint32_t msg_size; // 每条消息大小
uint32_t max_msgs; // 最大消息数
uint32_t head; // 读指针
uint32_t tail; // 写指针
uint32_t count; // 当前消息数
List waiting_send; // 等待发送的任务列表
List waiting_recv; // 等待接收的任务列表
} msg_queue_t;
这里的关键是两个等待列表。发送任务和接收任务各自排队，谁的条件满足了，谁就从阻塞态切换到就绪态。我在项目中遇到过一个问题：两个任务同时往同一个队列发消息，如果队列满了，它们都会挂到 waiting_send 列表里。等队列有空位了，谁先被唤醒？这取决于调度策略——通常是优先级高的先醒，同优先级就按 FIFO。
- 阻塞等待：任务挂起，直到条件满足才恢复。适合确定性场景。
- 超时等待：任务挂起，但最多等指定时间。超时后返回错误码，任务继续执行。适合容错场景。
你想想看，如果接收任务永远等不到消息，又没有超时机制，那这个任务就“死”在那了。所以我在实际项目中，接收端几乎必设超时，发送端则看情况。
4.2 生产者-消费者模型
这是消息队列最经典的应用场景。一个任务生产数据，一个或多个任务消费数据。中间用队列解耦。
我画了一张图，帮你理清这个模型的核心流程：
这个模型的好处很明显：生产者和消费者的速率可以不同。生产者快的时候，消息在队列里积压；消费者快的时候，队列变空。只要队列深度设计合理，系统就不会丢数据。
4.3 邮箱：轻量级的消息队列
邮箱和消息队列的区别，我用一个表格说清楚：
| 特性 | 消息队列 | 邮箱 |
|---|---|---|
| 容量 | 多条消息 | 通常只有1条 |
| 消息大小 | 固定大小（拷贝） | 指针或固定大小 |
| 阻塞机制 | 发送/接收均可阻塞 | 发送/接收均可阻塞 |
| 典型用途 | 数据流、生产者-消费者 | 事件通知、状态同步 |
| 内存开销 | 较大（缓冲区数组） | 很小（一个槽位） |
邮箱其实就是一个“单槽消息队列”。我常用邮箱来做任务间的事件通知。比如按键任务检测到按键，往邮箱里发一个“按键事件”结构体，UI任务阻塞等待这个邮箱。一来一回，代码非常简洁。
4.4 实战：串口数据分发
好了，理论讲完，咱们直接上实战。这是我在一个物联网网关项目里用过的方案：串口收到一帧数据，分发给多个处理任务。
先看整体架构：
// 定义消息结构
typedef struct {
uint8_t src_port; // 来源串口号
uint8_t *data; // 数据指针
uint16_t len; // 数据长度
uint32_t timestamp; // 时间戳
} serial_msg_t;
// 创建消息队列
msg_queue_t *g_serial_queue;
#define SERIAL_QUEUE_DEPTH 16
#define SERIAL_MSG_SIZE sizeof(serial_msg_t)
void serial_init(void) {
g_serial_queue = msg_queue_create(
SERIAL_QUEUE_DEPTH,
SERIAL_MSG_SIZE
);
// 检查创建是否成功
configASSERT(g_serial_queue != NULL);
}
串口中断服务程序里，只做最轻量的事：把数据搬到缓冲区，然后发消息给队列。注意，中断里不能阻塞，所以发送要用“不等待”模式。
// 串口中断回调（简化版）
void uart_isr_handler(uart_port_t port) {
uint8_t byte;
static uint8_t buf[256];
static uint16_t idx = 0;
while (uart_byte_available(port)) {
byte = uart_read_byte(port);
buf[idx++] = byte;
// 检测帧结束标志（比如换行符）
if (byte == '\n' || idx >= sizeof(buf)) {
// 构造消息
serial_msg_t msg;
msg.src_port = port;
msg.data = buf; // 注意：这里用静态缓冲区
msg.len = idx;
msg.timestamp = get_tick();
// 发送到队列，不等待
// 如果队列满，直接丢弃
msg_queue_send(g_serial_queue, &msg, 0);
idx = 0;
}
}
}
接下来是消费者任务。假设有三个任务：一个做协议解析，一个做日志记录，一个做数据转发。
// 协议解析任务
void task_protocol_parser(void *arg) {
serial_msg_t msg;
while (1) {
// 阻塞等待，超时设为100ms
if (msg_queue_recv(g_serial_queue, &msg, 100) == OK) {
// 解析协议帧
parse_protocol(msg.data, msg.len);
} else {
// 超时处理：可以做些低功耗操作
task_yield();
}
}
}
// 日志记录任务
void task_logger(void *arg) {
serial_msg_t msg;
while (1) {
if (msg_queue_recv(g_serial_queue, &msg, 50) == OK) {
save_to_log(msg.data, msg.len);
}
}
}
// 数据转发任务
void task_forwarder(void *arg) {
serial_msg_t msg;
while (1) {
if (msg_queue_recv(g_serial_queue, &msg, 200) == OK) {
forward_to_cloud(msg.data, msg.len);
}
}
}
三个任务都从同一个队列取消息。每条消息只会被其中一个任务取走。如果你想实现“广播”效果——每条消息发给所有消费者——那就得用发布-订阅模型，或者每个消费者有自己的队列，生产者往每个队列都发一份。
- 中断里只做最轻量的事：收数据、发消息。别在中断里做协议解析。
- 队列深度要够：串口速率高的时候，队列满了会丢帧。我一般按最大帧率的2倍来设。
- 消费者超时不能太长：否则队列积压，新数据进不来。建议超时设在50~200ms。
- 考虑优先级：协议解析任务优先级最高，日志任务最低。这样关键数据优先处理。
嗯，这里还要提一个我踩过的坑：消息队列里的数据拷贝问题。如果每条消息都拷贝完整的数据帧，内存开销会很大。我的做法是：消息里只放指针和长度，数据放在一个共享的环形缓冲区里。但这样就要处理缓冲区生命周期的问题——谁负责释放？我一般用引用计数，或者等所有消费者都读完再释放。
最后，聊一下低功耗场景下的消息队列。如果系统大部分时间在休眠，串口数据来了怎么办？我的方案是：串口中断只唤醒一个“收集中转”任务，这个任务收完一帧数据后，发消息给其他任务，然后自己继续休眠。其他任务收到消息后，处理完也继续休眠。这样整个系统的唤醒时间最短，功耗最低。