第4章 NuttX消息队列基础

第4章：NuttX消息队列基础

消息队列，说白了就是任务之间传递数据的“邮局”。

我刚开始接触RTOS时，总觉得信号量就能解决同步问题，干嘛还要消息队列？后来做一个传感器数据采集项目，发现多个任务要共享数据，用全局变量加信号量保护，代码越写越乱。嗯，这时候消息队列就派上用场了。

4.1 消息队列的概念

消息队列本质上是一个内核对象，它维护着一个先进先出的缓冲区。发送方把消息丢进去，接收方按顺序取出来。每个消息可以携带任意长度的数据。

我个人习惯把消息队列想象成一个管道——数据从一端流入，从另一端流出。任务之间通过这个管道解耦，发送方不用管谁在接收，接收方也不用管谁在发送。

核心特性：

异步通信：发送方不阻塞，接收方可以选择阻塞等待
数据缓冲：消息暂存在队列中，不丢失
多对多：多个任务可以往同一个队列发消息，多个任务也可以从同一个队列收消息
固定大小：每个消息的大小在创建时确定

你想想看，如果两个任务要交换数据，用全局变量加信号量保护，你得自己管理缓冲区、处理并发。用消息队列，这些脏活累活内核帮你干了。

4.2 消息队列的创建与删除

在NuttX中，消息队列的API很简洁。创建队列用mq_open，删除用mq_close和mq_unlink。

4.2.1 创建消息队列

#include <mqueue.h>

struct mq_attr attr;
mqd_t mqdes;

/* 设置队列属性 */
attr.mq_maxmsg  = 10;      /* 最大消息数 */
attr.mq_msgsize = 64;      /* 每个消息的最大字节数 */
attr.mq_flags   = 0;       /* 阻塞模式 */

/* 创建消息队列 */
mqdes = mq_open("/sensor_data", O_CREAT | O_RDWR, 0666, &attr);
if (mqdes == (mqd_t)-1) {
    printf("创建消息队列失败！\n");
    return -1;
}

这里要注意几个参数：

队列名：必须以/开头，比如/sensor_data。不同队列名字不能重复
最大消息数：决定了队列能缓存多少条消息。设太小容易满，设太大浪费内存
消息大小：每个消息的固定长度。我建议按实际数据结构的sizeof来设

我的经验：消息大小不要设得太大。64字节够用就别设256。每个消息都会预分配这块内存，设大了内存浪费很严重。我曾经在一个资源受限的项目中，把消息大小从128改成32，内存占用直接降了75%。

4.2.2 删除消息队列

/* 关闭队列描述符 */
mq_close(mqdes);

/* 删除队列（释放内核资源） */
mq_unlink("/sensor_data");

mq_close只是关闭当前任务的访问，mq_unlink才是真正删除队列。只有所有引用都关闭后，队列才会被销毁。

避坑指南：我曾经在项目里忘记调用mq_unlink，结果重启后队列还在，旧数据把新数据冲乱了。所以，队列用完一定要清理干净。

4.3 消息发送与接收

消息队列的核心操作就两个：发和收。NuttX提供了阻塞和非阻塞两种模式。

4.3.1 发送消息

#include <mqueue.h>

struct sensor_data {
    uint32_t timestamp;
    float    temperature;
    float    humidity;
};

struct sensor_data data;
mqd_t mqdes;

/* 准备数据 */
data.timestamp = get_time();
data.temperature = 25.6;
data.humidity = 68.3;

/* 发送消息（阻塞模式） */
int ret = mq_send(mqdes, (const char*)&data, sizeof(data), 0);
if (ret != 0) {
    printf("发送失败！队列可能已满\n");
}

mq_send的第四个参数是优先级，0表示普通优先级。如果队列满了，任务会阻塞直到有空间。

4.3.2 接收消息

struct sensor_data received;
unsigned int prio;

/* 接收消息（阻塞模式） */
ssize_t nbytes = mq_receive(mqdes, (char*)&received, sizeof(received), &prio);
if (nbytes == sizeof(received)) {
    printf("收到数据：温度=%.1f, 湿度=%.1f\n", 
           received.temperature, received.humidity);
} else {
    printf("接收失败！\n");
}

接收时，缓冲区大小必须大于等于创建队列时设定的mq_msgsize，否则会报错。

非阻塞模式：

在创建队列时设置O_NONBLOCK标志，或者用mq_setattr动态修改。非阻塞模式下，队列满时mq_send直接返回-1并设置errno为EAGAIN。

4.3.3 实际项目中的典型用法

我做一个数据采集系统时，用了三个任务：

采集任务：从传感器读数据，发到队列
处理任务：从队列取数据，做滤波和计算
上报任务：把处理结果通过串口发出去

/* 采集任务 */
void collector_task(int argc, char *argv[])
{
    mqd_t mq = mq_open("/data_queue", O_WRONLY);
    struct sensor_data data;
    
    while (1) {
        read_sensor(&data);
        mq_send(mq, (char*)&data, sizeof(data), 0);
        usleep(100000); /* 10Hz采样 */
    }
}

/* 处理任务 */
void processor_task(int argc, char *argv[])
{
    mqd_t mq = mq_open("/data_queue", O_RDONLY);
    struct sensor_data data;
    
    while (1) {
        mq_receive(mq, (char*)&data, sizeof(data), NULL);
        process_data(&data);
    }
}

你看，两个任务通过消息队列解耦，采集任务只管发，处理任务只管收。代码清晰，维护也方便。

4.4 消息队列的注意事项

问题	原因	解决方法
队列满导致阻塞	消费者处理速度跟不上生产者	增大队列深度，或使用非阻塞模式
消息丢失	接收缓冲区太小	确保接收缓冲区 ≥ mq_msgsize
内存泄漏	忘记mq_unlink	在任务退出时清理队列
优先级反转	高优先级任务等低优先级任务的消息	使用优先级继承或单独队列

我的建议：消息队列虽好，但别滥用。如果只是传递一个标志位，用信号量更轻量。消息队列适合传递结构化数据，比如传感器读数、网络包、命令帧等。

嗯，消息队列的基础就这些。说白了，它就是任务间传递数据的管道，用好了能让你的系统架构清晰很多。下一章我们聊聊信号量，看看它和消息队列怎么配合使用。

📚 课程目录