消息队列

📚 本节导读

学习时长: 约 30 分钟
难度级别: ⭐⭐⭐⭐☆
前置知识: 任务管理、信号量、事件

消息队列（Message Queue）是 OneOS 中用于任务间传输变长数据的核心通信机制。与信号量只能传递计数信号不同，消息队列可以传递任意大小的数据块，发送方和接收方各自持有独立的数据副本。

🎯 学习目标

• 理解消息队列的工作原理：数据拷贝与消息缓冲
• 掌握动态创建和静态创建消息队列的方法
• 掌握消息的发送和接收操作
• 理解紧急发送与普通发送的区别
• 能够实现生产者-消费者模式

一、消息队列概念

1.1 什么是消息队列

消息队列是一个消息缓冲池，内部维护一个环形缓冲区（或链表），用于存储变长消息。消息以数据拷贝的方式传递，发送方将数据拷贝到队列中，接收方从队列中拷贝出来。

┌─────────────────────────────────────────────────────────────────┐
│                     消息队列模型                                  │
│                                                                 │
│   ┌──────────┐   send(data, len)    ┌──────────────────────┐   │
│   │  任务 A   │ ──────────────────→ │   消息队列             │   │
│   │(生产者)  │    拷贝数据到队列     │  ┌───┬───┬───┬───┐   │   │
│   └──────────┘                      │  │msg│msg│msg│msg│   │   │
│                                     │  │ 1 │ 2 │ 3 │ 4 │   │   │
│   ┌──────────┐   recv(buf, len)    │  └───┴───┴───┴───┘   │   │
│   │  任务 B   │ ←────────────────── │                      │   │
│   │(消费者)  │    从队列拷贝数据     │  send_urgent 插入队首  │   │
│   └──────────┘                      │  ┌───┐               │   │
│                                     │  │紧急│→ 优先出队      │   │
│                                     │  └───┘               │   │
│                                     └──────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

1.2 消息队列 vs 其他 IPC

特性	消息队列	信号量	事件	邮箱
传递内容	变长数据块	计数（无数据）	32 位标志	指针值
数据拷贝	是（拷贝数据）	—	—	否（拷贝指针）
消息大小	每条消息独立大小	—	—	固定 os_ubase_t
缓冲能力	多条消息缓冲	计数器	位掩码	多条邮件缓冲
紧急发送	支持	—	—	不支持

1.3 内存池计算

静态创建消息队列时，需要预先分配内存池。内存池大小通过以下宏计算：

#define OS_MSGQUEUE_CALC_POOL_SIZE(msg_count, msg_size) \
    (sizeof(dummy_mq_msg_hdr_t) + OS_ALIGN_UP((msg_size), OS_ALIGN_SIZE)) * (msg_count)

每条消息包含一个消息头（dummy_mq_msg_hdr_t）和消息体，消息体按对齐大小向上取整，然后乘以消息总数。

二、API 详解

以下 API 定义在 kernel/include/os_mq.h 中：

#include <os_mq.h>

注意：使用消息队列需要在 oneos_config.h 中启用 OS_USING_MESSAGEQUEUE 宏。动态创建还需要 OS_USING_HEAP。

2.1 静态定义宏

#define OS_MSGQUEUE_DEFINE(name)                                  /* 定义消息队列控制块 */
#define OS_MSGQUEUE_POOL_DEFINE(name, msg_count, msg_size)        /* 定义消息队列内存池 */
#define OS_MSGQUEUE_POOL_ADDR(name)  ((void *)pool_##name)        /* 获取内存池地址 */
#define OS_MSGQUEUE_POOL_SIZE(name)  sizeof(pool_##name)          /* 获取内存池大小 */

2.2 动态创建

os_msgqueue_id os_msgqueue_create_dynamic(const char *name,
                                          os_size_t    msg_size,
                                          os_size_t    max_msgs);

参数	说明
name	消息队列名称
msg_size	每条消息的最大大小（字节）
max_msgs	消息队列能够容纳的最大消息数
返回值	成功返回消息队列 ID，失败返回 OS_NULL

需要启用 OS_USING_HEAP。

2.3 静态创建

os_msgqueue_id os_msgqueue_create_static(os_msgqueue_dummy_t *msgqueue,
                                         const char          *name,
                                         void                *msg_pool,
                                         os_size_t            msg_pool_size,
                                         os_size_t            msg_size);

参数	说明
msgqueue	消息队列控制块指针（OS_MSGQUEUE_DEFINE 定义的变量地址）
name	消息队列名称
msg_pool	消息队列内存池地址（OS_MSGQUEUE_POOL_ADDR 获取）
msg_pool_size	内存池大小（OS_MSGQUEUE_POOL_SIZE 获取）
msg_size	每条消息的最大大小
返回值	成功返回消息队列 ID，失败返回 OS_NULL

2.4 销毁消息队列

os_err_t os_msgqueue_destroy(os_msgqueue_id msgqueue_id);

2.5 发送消息

os_err_t os_msgqueue_send(os_msgqueue_id msgqueue_id,
                          const void    *buffer,
                          os_size_t      buff_size,
                          os_tick_t      timeout);

参数	说明
msgqueue_id	消息队列 ID
buffer	要发送的数据缓冲区指针
buff_size	数据大小（字节），不能超过消息队列的 msg_size
timeout	超时时间。队列满时等待空闲槽位的时间
返回值	OS_EOK 成功；OS_ETIMEOUT 超时；OS_EFULL 队列满

2.6 紧急发送

os_err_t os_msgqueue_send_urgent(os_msgqueue_id msgqueue_id,
                                 const void    *buffer,
                                 os_size_t      buff_size,
                                 os_tick_t      timeout);

与普通发送的区别：紧急消息插入队首，优先被接收者取出。普通消息插入队尾。

2.7 接收消息

os_err_t os_msgqueue_recv(os_msgqueue_id msgqueue_id,
                          void          *buffer,
                          os_size_t      buff_size,
                          os_tick_t      timeout,
                          os_size_t     *recv_msg_size);

参数	说明
msgqueue_id	消息队列 ID
buffer	接收数据的缓冲区指针
buff_size	接收缓冲区大小
timeout	超时时间
recv_msg_size	输出参数，返回实际接收到的消息大小
返回值	OS_EOK 成功；OS_ETIMEOUT 超时；OS_EEMPTY 队列空

2.8 查询函数

os_err_t    os_msgqueue_set_wake_type(os_msgqueue_id msgqueue_id, uint8_t wake_type);
os_err_t    os_msgqueue_reset(os_msgqueue_id msgqueue_id);
os_err_t    os_msgqueue_is_empty(os_msgqueue_id msgqueue_id, os_bool_t *empty_flag);
os_err_t    os_msgqueue_is_full(os_msgqueue_id msgqueue_id, os_bool_t *full_flag);
os_err_t    os_msgqueue_get_queue_depth(os_msgqueue_id msgqueue_id, uint16_t *msgqueue_depth);
os_err_t    os_msgqueue_get_max_msg_size(os_msgqueue_id msgqueue_id, os_size_t *max_msg_size);
os_err_t    os_msgqueue_get_used_msgs(os_msgqueue_id msgqueue_id, uint16_t *used_msgs);
os_err_t    os_msgqueue_get_unused_msgs(os_msgqueue_id msgqueue_id, uint16_t *unused_msgs);
const char *os_msgqueue_get_name(os_msgqueue_id msgqueue_id);

函数	说明
os_msgqueue_set_wake_type	设置唤醒策略：OS_MQ_WAKE_TYPE_PRIO 或 OS_MQ_WAKE_TYPE_FIFO
os_msgqueue_reset	重置消息队列，清空所有消息
os_msgqueue_is_empty	检查队列是否为空
os_msgqueue_is_full	检查队列是否已满
os_msgqueue_get_queue_depth	获取队列总容量（消息数）
os_msgqueue_get_max_msg_size	获取每条消息的最大大小
os_msgqueue_get_used_msgs	获取已使用的消息数
os_msgqueue_get_unused_msgs	获取剩余可用的消息数
os_msgqueue_get_name	获取消息队列名称

三、代码示例

3.1 动态创建：生产者-消费者

#include <oneos_config.h>
#include <os_task.h>
#include <os_mq.h>

#define MSG_SIZE    64
#define MAX_MSGS    10

typedef struct
{
    int  id;
    char data[MSG_SIZE - sizeof(int)];
} sensor_data_t;

/* 生产者任务 */
void sensor_task(void *parameter)
{
    os_msgqueue_id mq_id = (os_msgqueue_id)parameter;
    sensor_data_t  sensor_msg;
    static int     msg_id = 0;

    while (1)
    {
        /* 采集传感器数据 */
        sensor_msg.id = msg_id++;
        snprintf(sensor_msg.data, sizeof(sensor_msg.data),
                 "sensor_value_%d", msg_id);

        /* 发送到消息队列 */
        os_err_t ret = os_msgqueue_send(mq_id, &sensor_msg,
                                        sizeof(sensor_msg), OS_WAIT_FOREVER);
        if (ret == OS_EOK)
            printf("Sensor: sent msg %d\n", sensor_msg.id);

        os_task_sleep(500);
    }
}

/* 消费者任务 */
void processor_task(void *parameter)
{
    os_msgqueue_id mq_id = (os_msgqueue_id)parameter;
    sensor_data_t  recv_msg;
    os_size_t      recv_size;

    while (1)
    {
        os_err_t ret = os_msgqueue_recv(mq_id, &recv_msg,
                                        sizeof(recv_msg),
                                        OS_WAIT_FOREVER,
                                        &recv_size);
        if (ret == OS_EOK)
        {
            printf("Processor: received msg %d, size=%d, data=%s\n",
                   recv_msg.id, (int)recv_size, recv_msg.data);
        }
    }
}

void mq_dynamic_example(void)
{
    os_msgqueue_id mq_id;

    /* 动态创建消息队列 */
    mq_id = os_msgqueue_create_dynamic("sensor_mq", MSG_SIZE, MAX_MSGS);
    if (mq_id == OS_NULL)
    {
        printf("Failed to create message queue\n");
        return;
    }

    /* 创建生产者和消费者任务 */
    /* ... os_task_create(...) */
}

3.2 静态创建

#define MY_MSG_COUNT 8
#define MY_MSG_SIZE  32

/* 静态定义消息队列控制块 */
OS_MSGQUEUE_DEFINE(g_static_mq);

/* 静态定义消息队列内存池 */
OS_MSGQUEUE_POOL_DEFINE(g_static_mq, MY_MSG_COUNT, MY_MSG_SIZE);

void mq_static_example(void)
{
    os_msgqueue_id mq_id;

    mq_id = os_msgqueue_create_static(
        &g_static_mq,                              /* 控制块 */
        "static_mq",                                /* 名称 */
        OS_MSGQUEUE_POOL_ADDR(g_static_mq),        /* 内存池地址 */
        OS_MSGQUEUE_POOL_SIZE(g_static_mq),        /* 内存池大小 */
        MY_MSG_SIZE                                 /* 消息大小 */
    );

    if (mq_id == OS_NULL)
    {
        printf("Failed to create static message queue\n");
        return;
    }

    /* 发送消息 */
    char msg[] = "Hello from static MQ!";
    os_msgqueue_send(mq_id, msg, sizeof(msg), OS_NO_WAIT);
}

3.3 紧急发送 vs 普通发送

void urgent_send_example(os_msgqueue_id mq_id)
{
    char normal_msg[] = "Normal message";
    char urgent_msg[] = "URGENT message";

    /* 普通发送：插入队尾 */
    os_msgqueue_send(mq_id, normal_msg, sizeof(normal_msg), OS_NO_WAIT);

    /* 紧急发送：插入队首，优先被接收 */
    os_msgqueue_send_urgent(mq_id, urgent_msg, sizeof(urgent_msg), OS_NO_WAIT);

    /* 接收时，紧急消息先被取出 */
    char recv_buf[64];
    os_size_t recv_size;

    os_msgqueue_recv(mq_id, recv_buf, sizeof(recv_buf), OS_NO_WAIT, &recv_size);
    printf("First received: %s\n", recv_buf);  /* 输出 "URGENT message" */
}

3.4 查询消息队列状态

void mq_status_example(os_msgqueue_id mq_id)
{
    os_bool_t empty_flag;
    os_bool_t full_flag;
    uint16_t  depth;
    uint16_t  used_msgs;
    uint16_t  unused_msgs;

    os_msgqueue_is_empty(mq_id, &empty_flag);
    os_msgqueue_is_full(mq_id, &full_flag);
    os_msgqueue_get_queue_depth(mq_id, &depth);
    os_msgqueue_get_used_msgs(mq_id, &used_msgs);
    os_msgqueue_get_unused_msgs(mq_id, &unused_msgs);

    printf("MQ Status: empty=%d, full=%d, capacity=%d, used=%d, free=%d\n",
           empty_flag, full_flag, depth, used_msgs, unused_msgs);
}

📝 本节小结

消息队列是 OneOS 中传递变长数据的核心通信机制，关键要点：

1. 数据拷贝：消息队列以数据拷贝方式传递，发送方和接收方各自持有独立数据副本
2. 动态和静态创建：动态创建灵活（需堆），静态创建确定性强（无碎片）
3. 内存池计算：OS_MSGQUEUE_CALC_POOL_SIZE(msg_count, msg_size) 帮助计算静态内存池大小
4. 紧急发送：send_urgent 将消息插入队首，优先被接收
5. 超时机制：发送和接收都支持超时，避免死等
6. 消息队列不能在中断上下文中使用（与信号量不同）

🔗 相关链接

• 4.3 任务同步与通信
• 4.3.5 邮箱
• 4.3.1 信号量
• OneOS 内核源码 os_mq.h