消息队列

📚 本节导读

学习时长: 约 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 消息队列使用场景

一句话核心：消息队列适用于任务之间异步、有缓冲地传递一段数据（变长报文/命令/采集值）的场景。当通信的"内容"不只是一个标志/计数，而是一段具体数据时，就该用消息队列。

先看它与其他机制的本质区别，这直接决定了适用场景：

机制	传递的是什么	能否携带数据	有界性
信号量	计数（1 个整型）	否	有界（最大计数值）
事件	事件标志位（bitmask）	否	有界（32 位）
互斥锁	所有权	否	单资源
消息队列	消息块（带长度、带内容）	是，可携带任意数据	有界（池大小、消息数）

典型场景一：生产者-消费者 / 数据采集转存（最典型）

任务 A 采集/产生数据，任务 B 消费/处理数据，中间用队列解耦，避免双方直接耦合或长时间占用对方。

采集任务A ──os_msgqueue_send──► [消息队列] ──os_msgqueue_recv──► 处理任务B
   （高频生产）                    （缓冲n条）                     （消费/处理）

为什么用消息队列：生产者和消费者速率可能不同，队列提供缓冲（max_msgs 条消息），生产快时消息排队、消费快时阻塞等待，双方不互相死等。典型如传感器数据采集 → 处理、串口收包 → 协议解析。

典型场景二：任务间传递"结构化命令/数据包"

当要传递的是带语义的数据包（命令码 + 参数、协议帧、报文）时，信号量/事件只能传标志传不了数据，必须用消息队列。且每条消息有自己的 msg_len（可变长度），适合传长度可变的报文。典型如：命令任务向工作任务发"执行某操作 + 参数"；网络接收任务把解析好的协议帧发给上层业务任务；输入任务把按键事件（键值 + 按下/抬起）发给 UI 任务。

典型场景三：中断通知任务处理（有数据要携带时）

如果中断只发"有事件"的信号，用信号量/事件即可；但如果中断还需要把采集到的数据交给任务，则用消息队列（中断里 os_msgqueue_send，注意 timeout 必须用 OS_NO_WAIT，见源码 809 行的 OS_ASSERT 限制）。

中断ISR ──os_msgqueue_send(..., OS_NO_WAIT)──► [队列] ──► 处理任务

为什么用消息队列：中断里不能阻塞，用 OS_NO_WAIT 尝试入队（满则返回 OS_FULL），把数据安全地从中断上下文搬运到任务上下文。

典型场景四：多任务向单一消费者汇聚（Fan-in）

多个任务给一个任务发消息，队列天然是多生产者/单消费者模型：

任务P1 ──┐
任务P2 ──┼──► [消息队列] ──► 消费者任务C
任务P3 ──┘

为什么用消息队列：队列本身有锁（mq->lock/临界区）保护，多任务并发 send 是安全的，无需使用者额外加锁；recv 侧单消费者按优先级/FIFO 依次取。

典型场景五：三种配置速览（1 对 1 / N 对 1 / N 对 N）——多接收者到底什么时候出现

消息队列的"生产-消费"关系可以按收发任务数量分为三种配置。"接收者等待队列里出现多个接收者"只在 N 对 N（多生产者/多消费者）且队列为空时才出现，下面逐一结合实际工作情况举例。

配置	典型实际场景	接收者等待队列是否会出现多个接收者
1 对 1	传感器采样 → 显示；中断 → 任务	否（始终只有 1 个接收者）
N 对 1	多传感器汇聚 → 中央控制；多外设 → UI	否（接收方唯一）
N 对 N	工作线程池、网络多核分发	是（多任务竞争同一队列，队列空时都阻塞挂入）

① 一个发送 → 一个接收（1 对 1，最常用）

采样任务 S ──os_msgqueue_send(温度值)──► [消息队列 MQ] ──os_msgqueue_recv──► 显示任务 D
（每 100ms 采一次）                        （缓存最近若干条）                  （更新屏显/告警）

实际工作情况：

• 采样任务 S 每 100ms 采集一次温度（如 25.3℃），打包成一条消息 send 进队列；
• 显示任务 D 循环 recv，取到就更新 LCD / 触发高温告警；
• 若 D 处理慢（LCD 刷新耗 300ms），队列缓存多条温度，D 落后时可只显示最新（丢弃旧消息）；
• 双方若都用 OS_WAIT_FOREVER，则 S 队列满时阻塞、D 队列空时阻塞，生产-消费节奏天然同步。

变体——中断到任务（单向）：按键中断把键值发给 UI 任务，ISR 里 send 必须用 OS_NO_WAIT（中断不能阻塞），用户快速连按而 UI 忙时队列缓冲，不丢按键（只要没满）。

② 多个发送 → 一个接收（N 对 1，多源汇聚）

温度采集任务 ──┐
湿度采集任务 ──┼── os_msgqueue_send ──► [消息队列 MQ] ── os_msgqueue_recv ──► 中央控制任务 C
气压采集任务 ──┘

实际工作情况：

• 三个采集任务各自独立运行，采集完就把"传感器 ID + 数据"打包 send；
• 中央任务 C 统一 recv，根据消息里的传感器 ID 区分数据来源，分别记录/控制；
• 三生产者并发 send，队列内部有锁（mq->lock/临界区）保护，C 无需额外加锁；
• 若 C 处理慢，队列缓存，C 按 wake_type（PRIO/FIFO）依次消费。此配置接收者唯一，不会出现多个接收者。

③ 多个发送 → 多个接收（N 对 N，负载均衡）——多接收者的真正来源

主控任务 M ──send(指令1,2,3...)──► [消息队列 MQ] ─┬─► 工作任务 W1
                                              ├─► 工作任务 W2
                                              └─► 工作任务 W3
每个 W 都执行 os_msgqueue_recv（队列空时阻塞进接收者等待队列）

时间线（为什么会出现多个接收者）：

时刻T1：M 下发 5 条指令（1~5）
        W1 recv→取指令1，W2 recv→取指令2，W3 recv→取指令3
时刻T2：队列还剩 4、5（未空）；W2 最快处理完 → recv→取指令4
时刻T3：队列空了，此时 W1、W3 处理完又来 recv
        W1 → 队列空，阻塞挂进 recv_task_list_head
        W3 → 队列空，也阻塞挂进来（等待队列：W1 → W3）
时刻T4：M 又 send 指令6 → 唤醒队首 W1，W1 取走指令6
时刻T5：再 send 指令7 → 唤醒 W3，W3 取走指令7

实际工作情况：

• 负载均衡：谁空闲谁取，快的工作线程取更多条指令，天然按处理速度分配；
• 提高吞吐：指令处理（如电机控制）耗时长，多线程并行才能跟上主控下发速度；单线程会堆积 OS_FULL；
• 队列做缓冲：主控发得快时指令排队，工作线程慢时不阻塞主控（只要队列没满）；
• 每次只唤醒队首一个：os_msgqueue_send 用 os_list_first_entry 取队首接收者唤醒，一条消息给一个接收者，其余继续阻塞等下一条。

网络报文分发变体：网卡接收任务高速 send，多个处理核任务 recv 并行解包/转发；某核处理慢（如丢包重传）不拖累其他核，高流量时队列缓冲、低流量时处理核阻塞省电。

什么时候不该用消息队列（选型边界）

场景	该用什么	为什么
只通知"发生了事件"，无数据	事件 / 信号量	用消息队列浪费内存（消息块池）
只需传 1 个标志位/计数值	信号量	更轻量
需要传 32 位以内的多个标志位	事件	一次传多个 bit
需要"排他访问共享资源"	互斥锁	消息队列不做所有权/优先级继承
实时性要求极高、需保证不丢消息	需谨慎	队列满时 OS_NO_WAIT 会返回 OS_FULL 丢消息；OS_WAIT_FOREVER 会阻塞

结合 OneOS 源码的使用注意点

1. 有界性：队列有 queue_depth（max_msgs）、max_msg_size，消息块池大小固定。设计时估算峰值消息数——池太小会返回 OS_FULL，太大浪费内存。
2. 可变消息长度：os_msgqueue_send 需传 buff_size，os_msgqueue_recv 需保证 buff_size >= msg->msg_len，否则接收者会丢弃该消息（返回 OS_NOMEM，见 recv 源码 875-885 行）——接收 buffer 一定要开够。
3. 中断上下文：中断里 send 必须用 OS_NO_WAIT（源码 assert 限制了等待条件）。
4. 唤醒策略：wake_type 可选 OS_MQ_WAKE_TYPE_PRIO（按优先级唤醒）或 FIFO，适合有优先级要求的场景。

1.4 内存池计算

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

#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）和消息体，消息体按对齐大小向上取整，然后乘以消息总数。

1.5 消息队列控制块（os_msgqueue_t）字段详解

消息队列的内部控制块 os_msgqueue_t（即 struct os_msgqueue）定义在 kernel/source/os_prototypes.h 中：

struct os_msgqueue
{
    void        *msg_pool;       /* Point to message pool. */
    os_mq_msg_t *msg_queue_head; /* Point to first entry of message queue. */
    os_mq_msg_t *msg_queue_tail; /* Point to last entry of message queue. */
    os_mq_msg_t *msg_queue_free; /* Point to first entry of free message resource. */

    os_list_node_t send_task_list_head; /* Sender tasks blocked on this messages queue. */
    os_list_node_t recv_task_list_head; /* Receiver tasks blocked on this messages queue. */

    os_list_node_t resource_node; /* Node in resource list */

    os_size_t max_msg_size; /* Max message size of each message. */
    uint16_t  queue_depth;  /* Maximum number of messages that can be put into message queue. */
    uint16_t  used_msgs;    /* Number of messages queued. */

    uint8_t object_inited;     /* Indicates whether object is inited or deinited */
    uint8_t object_alloc_type; /* Indicates whether object is allocated dynamically or statically */
    uint8_t wake_type;         /* The type to wake up blocking tasks */

    char name[OS_NAME_MAX + 1]; /* Message queue name. */

    os_spinlock_t lock; /* Spin lock used under SMP */
};
typedef struct os_msgqueue os_msgqueue_t;

各字段含义如下：

字段	类型	说明
msg_pool	void *	消息内存池指针：存放所有消息的内存区域
msg_queue_head	os_mq_msg_t *	队首指针：指向消息队列中第一条消息
msg_queue_tail	os_mq_msg_t *	队尾指针：指向消息队列中最后一条消息
msg_queue_free	os_mq_msg_t *	空闲消息指针：指向第一个空闲消息资源
send_task_list_head	os_list_node_t	发送者等待队列：因队列满而被阻塞的发送任务挂在这里
recv_task_list_head	os_list_node_t	接收者等待队列：因队列空而被阻塞的接收任务挂在这里
resource_node	os_list_node_t	资源链表节点：把消息队列挂到消息队列资源链表（gs_os_mq_resource_list_head）
max_msg_size	os_size_t	单条消息最大大小（字节）
queue_depth	uint16_t	队列容量：最多能容纳的消息数
used_msgs	uint16_t	已用消息数：当前队列中已存入的消息数
object_inited	uint8_t	内核对象初始化状态（OS_KOBJ_INITED / OS_KOBJ_DEINITED）
object_alloc_type	uint8_t	分配类型：静态 / 动态
wake_type	uint8_t	唤醒策略：OS_MQ_WAKE_TYPE_PRIO / OS_MQ_WAKE_TYPE_FIFO
name	char[OS_NAME_MAX+1]	消息队列名称
lock	os_spinlock_t	自旋锁（仅 SMP 编译）

关键字段与消息队列核心概念的对应：

消息队列概念	对应字段	说明
消息缓冲	msg_pool、msg_queue_head/tail/free	内存池存放消息，队首/队尾/空闲指针管理消息链表
容量与用量	queue_depth、used_msgs	队列容量 / 当前已用消息数（判空判满的依据）
满时阻塞发送者	send_task_list_head	队列满时，发送任务阻塞挂到这里
空时阻塞接收者	recv_task_list_head	队列空时，接收任务阻塞挂到这里
系统管理	resource_node	登记到全局消息队列资源链表，可查询/防重复

说明：用户代码不应直接访问这些内部字段，而是通过 os_mq.h 提供的 API 操作。此处的字段解析用于帮助理解消息队列的底层工作原理。

1.6 消息队列与多个链表的整体关系图

消息队列涉及三个链表：

链表	链表头	属于谁	存的内容
消息队列资源链表	gs_os_mq_resource_list_head	消息队列模块（全局 1 个）	所有已创建的消息队列（挂 resource_node）
发送者等待队列	send_task_list_head（队列内字段）	消息队列自己	队列满时阻塞的发送任务
接收者等待队列	recv_task_list_head（队列内字段）	消息队列自己	队列空时阻塞的接收任务

整体关系图：

【链表 1】消息队列资源链表（全局 1 个）
gs_os_mq_resource_list_head
     │
     ▼
  ┌──────────┐   ┌──────────┐   ┌──────────┐
  │  队列A    │──►│  队列B    │──►│  队列C    │  ← 通过各队列的 resource_node 串联
  └──────────┘   └──────────┘   └──────────┘

【链表 2】发送者等待队列（每个队列 1 个，存于队列内）
队列C 的 send_task_list_head  →  发送任务 S1 → 发送任务 S2
     │
     │  （队列满时，等待发送的 S1/S2 被 os_msgqueue_send 阻塞挂到这里）

【链表 3】接收者等待队列（每个队列 1 个，存于队列内）
队列C 的 recv_task_list_head  →  接收任务 R1 → 接收任务 R2
     │
     │  （队列空时，等待接收的 R1/R2 被 os_msgqueue_recv 阻塞挂到这里）

阶段一：队列创建后、无任务阻塞

  队列 C（刚创建）
  ┌──────────────────────────────┐
  │ resource_node ═══► 已挂到消息队列资源链表   （创建时 os_list_add_tail 挂入）
  │ send_task_list_head ═══► 空（无发送者阻塞）   │
  │ recv_task_list_head ═══► 空（无接收者阻塞）   │
  │ used_msgs = 0（队列空）                      │
  └──────────────────────────────┘

阶段二：队列空，且有接收者阻塞；队列满，且有发送者阻塞

  队列 C
  ┌──────────────────────────────┐
  │ resource_node ═══► 仍在资源链表（不变）        │
  │ send_task_list_head ═══► 发送任务 S1（队列满时）│
  │ recv_task_list_head ═══► 接收任务 R1（队列空时）│
  │ used_msgs = queue_depth（满）或 0（空）        │
  └──────────────────────────────┘

核心关系表：

链表	属于谁	挂哪个节点	何时挂入
消息队列资源链表（gs_os_mq_resource_list_head）	消息队列模块（全局 1 个）	resource_node	创建时
发送者等待队列（send_task_list_head）	消息队列自己	阻塞的发送任务	发送时队列满（os_msgqueue_send）
接收者等待队列（recv_task_list_head）	消息队列自己	阻塞的接收任务	接收时队列空（os_msgqueue_recv）

一句话：

• resource_node 让消息队列"被系统管理"（全局可见，可查询/防重复）。
• send_task_list_head 记录"哪些发送者因队列满在等空闲槽位"（队列有空间时唤醒）。
• recv_task_list_head 记录"哪些接收者因队列空在等消息"（有新消息时唤醒）。

说明：与互斥锁（有所有权）不同，消息队列没有所有权概念，但有两个等待队列（发送者 + 接收者），这是消息队列的特殊之处。

二、API 详解

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

#include <os_mq.h>

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

2.1 静态定义宏

静态创建消息队列时，需要用户提供控制块和内存池。这些通过 os_mq.h 提供的宏来定义。完整宏定义（kernel/include/os_mq.h，与项目一致）：

/* 定义消息队列控制块 */
#define OS_MSGQUEUE_DEFINE(name) os_msgqueue_dummy_t name

/*
 * OS_MQ_CALC_POOL_SIZE is used to calc the pool size
 * OS_MQ_POOL_DEFINE is used to define message queue's memory pool and the address is aligned.
 * OS_MQ_POOL_ADDR is used to get address of message queue's memory pool.
 * OS_MQ_POOL_SIZE is used to get size of message queue's memory pool.
 */
#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)
#define OS_MSGQUEUE_POOL_DEFINE(name, msg_count, msg_size)                                                             \
    OS_ALIGN(OS_ALIGN_SIZE)                                                                                            \
    uint8_t pool_##name[(sizeof(dummy_mq_msg_hdr_t) + OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE)) * msg_count]
#define OS_MSGQUEUE_POOL_ADDR(name) ((void *)pool_##name)
#define OS_MSGQUEUE_POOL_SIZE(name) sizeof(pool_##name)

各宏详细解释：

OS_MSGQUEUE_DEFINE(name) —— 定义消息队列控制块

#define OS_MSGQUEUE_DEFINE(name) os_msgqueue_dummy_t name

• 展开为一条变量声明：os_msgqueue_dummy_t name。
• os_msgqueue_dummy_t 是给用户看的占位类型（typedef struct dummy_msgqueue），它的内存布局、大小与内核真正的 os_msgqueue_t 完全一致。
• 用户用它定义一个全局变量 name，这块静态内存将来被内核当作真正的消息队列控制块使用。
• 用法示例：

OS_MSGQUEUE_DEFINE(my_mq);   /* 展开为：os_msgqueue_dummy_t my_mq; */

OS_MSGQUEUE_CALC_POOL_SIZE(msg_count, msg_size) —— 计算内存池大小

#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)

• 公式：每条消息的大小（消息头 + 对齐后的消息体）× 消息数。
  • sizeof(dummy_mq_msg_hdr_t)：消息头大小（每条消息都带一个消息头）。
  • OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE)：消息体大小按对齐边界向上取整。
  • × msg_count：乘以消息总数，得到整个内存池大小。
• 这个宏用于计算内存池需要多大，供 OS_MSGQUEUE_POOL_DEFINE 使用。

OS_MSGQUEUE_POOL_DEFINE(name, msg_count, msg_size) —— 定义内存池

#define OS_MSGQUEUE_POOL_DEFINE(name, msg_count, msg_size)                                                             \
    OS_ALIGN(OS_ALIGN_SIZE)                                                                                            \
    uint8_t pool_##name[(sizeof(dummy_mq_msg_hdr_t) + OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE)) * msg_count]

• 展开为一个对齐的全局数组：OS_ALIGN(OS_ALIGN_SIZE) uint8_t pool_name[...]。
• OS_ALIGN(OS_ALIGN_SIZE)：数组按对齐边界对齐（保证内存池起始地址满足对齐要求）。
• pool_##name：宏拼接，生成数组名 pool_my_mq 这样的全局数组。
• 数组大小就是 OS_MSGQUEUE_CALC_POOL_SIZE(msg_count, msg_size) 算出的内存池大小。
• 用法示例：

OS_MSGQUEUE_POOL_DEFINE(my_mq, 8, 32);   /* 展开为：对齐的 uint8_t pool_my_mq[...] 数组，容量 8 条×32 字节 */

OS_MSGQUEUE_POOL_ADDR(name) / OS_MSGQUEUE_POOL_SIZE(name) —— 获取内存池地址/大小

#define OS_MSGQUEUE_POOL_ADDR(name) ((void *)pool_##name)
#define OS_MSGQUEUE_POOL_SIZE(name) sizeof(pool_##name)

• OS_MSGQUEUE_POOL_ADDR(name)：返回内存池数组的起始地址（(void *)pool_name）。
• OS_MSGQUEUE_POOL_SIZE(name)：返回内存池数组的大小（sizeof(pool_name)）。
• 这两个宏用于 os_msgqueue_create_static() 的参数：

os_msgqueue_create_static(&my_mq,             /* 控制块地址（OS_MSGQUEUE_DEFINE） */
                          "my_mq",
                          OS_MSGQUEUE_POOL_ADDR(my_mq),   /* 内存池地址 */
                          OS_MSGQUEUE_POOL_SIZE(my_mq),   /* 内存池大小 */
                          32);                            /* 消息大小 */

静态定义宏使用流程：

① OS_MSGQUEUE_DEFINE(my_mq)         → 定义控制块变量 my_mq
② OS_MSGQUEUE_POOL_DEFINE(my_mq, 8, 32)  → 定义内存池数组 pool_my_mq
③ os_msgqueue_create_static(&my_mq, "my_mq",
                             OS_MSGQUEUE_POOL_ADDR(my_mq),
                             OS_MSGQUEUE_POOL_SIZE(my_mq), 32)

注意：静态创建不需要堆（控制块和内存池都是编译期分配的全局变量），适合资源受限、不使用动态内存的 MCU 场景。

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。

完整源码（kernel/source/os_mq.c，与项目一致）：

os_msgqueue_id os_msgqueue_create_dynamic(const char *name, os_size_t msg_size, os_size_t max_msgs)
{
    os_msgqueue_t *mq;
    void          *msg_pool;
    os_size_t      msg_pool_size;
    os_size_t      align_msg_size;
    os_err_t       ret;

    OS_ASSERT(msg_size > 0U);
    OS_ASSERT(max_msgs > 0U);
    OS_ASSERT(OS_FALSE == os_is_irq_active());

    align_msg_size = OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE);
    msg_pool_size  = max_msgs * (align_msg_size + sizeof(os_mq_msg_hdr_t));
    mq             = (os_msgqueue_t *)OS_KERNEL_MALLOC(sizeof(os_msgqueue_t));
    msg_pool       = OS_KERNEL_MALLOC_ALIGN(OS_ALIGN_SIZE, msg_pool_size);

    if ((OS_NULL == mq) || (OS_NULL == msg_pool))
    {
        OS_KERN_LOG(KERN_ERROR, MQ_TAG, "Malloc failed, mq(%p), msg_pool(%p)", mq, msg_pool);

        if (OS_NULL != mq)
        {
            OS_KERNEL_FREE(mq);
            mq = OS_NULL;
        }

        if (OS_NULL != msg_pool)
        {
            OS_KERNEL_FREE(msg_pool);
            msg_pool = OS_NULL;
        }

        ret = OS_NOMEM;
    }
    else
    {
        ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, align_msg_size, OS_KOBJ_ALLOC_TYPE_DYNAMIC);
        if (OS_SUCCESS == ret)
        {
            os_spin_lock(&gs_os_mq_resource_list_lock);
            os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
            os_spin_unlock(&gs_os_mq_resource_list_lock);

            mq->object_inited = OS_KOBJ_INITED;
        }
        else
        {
            OS_KERNEL_FREE(mq);
            OS_KERNEL_FREE(msg_pool);

            mq       = OS_NULL;
            msg_pool = OS_NULL;
        }
    }

    return OS_TYPE_CONVERT(os_msgqueue_id, mq);
}

分步详解：

第一步：参数校验（条件：msg_size、max_msgs 非 0，且不能在中断上下文）—— 不满足则断言崩溃

    OS_ASSERT(msg_size > 0U);
    OS_ASSERT(max_msgs > 0U);
    OS_ASSERT(OS_FALSE == os_is_irq_active());

进入本步的条件：调用 os_msgqueue_create_dynamic() 时无条件执行（本步是函数入口的硬性校验，不满足条件不会返回错误码，而是直接断言崩溃）。

详细解释：

• OS_ASSERT(msg_size > 0U)：条件 msg_size > 0——消息大小必须非 0，否则消息体长度为 0 无意义。
• OS_ASSERT(max_msgs > 0U)：条件 max_msgs > 0——队列容量必须非 0，否则无法容纳任何消息。
• OS_ASSERT(OS_FALSE == os_is_irq_active())：条件 os_is_irq_active() == OS_FALSE——不能在中断上下文创建，因为 OS_KERNEL_MALLOC 需要进程上下文（可能阻塞/调度）。

关键点：这三条是 OS_ASSERT（断言），失败即崩溃而非返回错误码。调用方必须保证参数合法，不能靠返回值兜底。

第二步：计算内存池大小并分配内存（条件：第一步断言通过）—— 分配控制块 + 对齐内存池

    align_msg_size = OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE);
    msg_pool_size  = max_msgs * (align_msg_size + sizeof(os_mq_msg_hdr_t));
    mq             = (os_msgqueue_t *)OS_KERNEL_MALLOC(sizeof(os_msgqueue_t));
    msg_pool       = OS_KERNEL_MALLOC_ALIGN(OS_ALIGN_SIZE, msg_pool_size);

进入本步的条件：第一步三条断言全部通过（msg_size > 0、max_msgs > 0、非中断上下文），函数继续执行到内存分配。

详细解释：

• OS_ALIGN_UP(msg_size, OS_ALIGN_SIZE)：条件保证消息大小按对齐边界向上取整（如 msg_size=30、OS_ALIGN_SIZE=4 → 取整为 32），确保每条消息体起始地址对齐。
• msg_pool_size = max_msgs * (对齐后的消息大小 + 消息头大小)：算出整个内存池大小（每条消息含消息头 os_mq_msg_hdr_t）。
• OS_KERNEL_MALLOC(sizeof(os_msgqueue_t))：分配控制块 mq 的内存。
• OS_KERNEL_MALLOC_ALIGN(OS_ALIGN_SIZE, msg_pool_size)：分配按对齐边界对齐的内存池 msg_pool。

关键点：此时 mq 和 msg_pool 都可能分配失败（返回 OS_NULL），所以下一步必须检查这两个指针——这是下一步的条件来源。

第三步：分配失败处理 —— 回滚（条件：mq == OS_NULL 或 msg_pool == OS_NULL）—— 释放已分配部分，返回 OS_NOMEM

    if ((OS_NULL == mq) || (OS_NULL == msg_pool))
    {
        OS_KERN_LOG(KERN_ERROR, MQ_TAG, "Malloc failed, mq(%p), msg_pool(%p)", mq, msg_pool);

        if (OS_NULL != mq)
        {
            OS_KERNEL_FREE(mq);
            mq = OS_NULL;
        }
        if (OS_NULL != msg_pool)
        {
            OS_KERNEL_FREE(msg_pool);
            msg_pool = OS_NULL;
        }
        ret = OS_NOMEM;
    }

进入本步的条件：mq == OS_NULL 或 msg_pool == OS_NULL（第二步的 OS_KERNEL_MALLOC 或 OS_KERNEL_MALLOC_ALIGN 至少一个失败，即内存不足）。

详细解释：

• 打印错误日志 Malloc failed，报告 mq 和 msg_pool 的实际地址（哪个是 NULL 一目了然）。
• 子分支 A：条件 mq != OS_NULL——控制块分配成功但内存池失败，则 OS_KERNEL_FREE(mq) 释放控制块，并把 mq 置 OS_NULL（防误用）。
• 子分支 B：条件 msg_pool != OS_NULL——内存池分配成功但控制块失败，则 OS_KERNEL_FREE(msg_pool) 释放内存池，并置 OS_NULL。
• 最后 ret = OS_NOMEM（内存不足错误码）。

关键点：这是典型的部分成功回滚——两个资源只分配成功一个时，必须把已成功的那个也释放，避免内存泄漏；且 mq、msg_pool 都置空，保证返回的队列 ID 是 OS_NULL。

第四步：初始化并加入资源链表（条件：分配全部成功，ret == OS_SUCCESS）—— 初始化 + 登记资源链表 + 标记已初始化

    else
    {
        ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, align_msg_size, OS_KOBJ_ALLOC_TYPE_DYNAMIC);
        if (OS_SUCCESS == ret)
        {
            os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
            mq->object_inited = OS_KOBJ_INITED;
        }
        else
        {
            OS_KERNEL_FREE(mq);
            OS_KERNEL_FREE(msg_pool);
            mq       = OS_NULL;
            msg_pool = OS_NULL;
        }
    }

进入本步的条件：第三步的条件不成立，即 mq != OS_NULL 且 msg_pool != OS_NULL（两个资源都分配成功），走到 else 分支。

详细解释：

• _k_mq_init(mq, name, msg_pool, msg_pool_size, align_msg_size, OS_KOBJ_ALLOC_TYPE_DYNAMIC)：初始化控制块各字段，返回 ret。
• 子分支 A：条件 ret == OS_SUCCESS（初始化成功）：
  • os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node)：把队列的 resource_node 登记进消息队列资源链表（挂到尾部，让系统全局可查）；
  • mq->object_inited = OS_KOBJ_INITED：标记该队列已初始化（后续 send/recv 前的断言会检查此字段）。
• 子分支 B：条件 ret != OS_SUCCESS（初始化失败）：释放控制块和内存池，mq、msg_pool 置 OS_NULL（回滚，不留下半初始化对象）。

关键点：资源链表登记与 object_inited 标记是成对出现的——只有登记成功才标记"已初始化"，否则整体回滚。这样其他 API（如 os_msgqueue_send）的 OS_ASSERT(OS_KOBJ_INITED == mq->object_inited) 才能正确判断队列是否可用。

最终返回值：return OS_TYPE_CONVERT(os_msgqueue_id, mq)——成功返回队列 ID；失败时 mq 已被置 OS_NULL，故返回 OS_NULL。

使用示例：

os_msgqueue_id mq_id = os_msgqueue_create_dynamic("mq", 64, 10);   /* 每条 64 字节，最多 10 条 */
if (mq_id == OS_NULL)
{
    os_kprintf("create failed\r\n");   /* 检查返回值 */
    return;
}

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

完整源码（kernel/source/os_mq.c，与项目一致）：

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)
{
    os_msgqueue_t  *iter_mq;
    os_list_node_t *pos;
    os_err_t        ret;
    os_msgqueue_t  *mq;

    OS_ASSERT(OS_NULL != msgqueue);
    OS_ASSERT(OS_NULL != msg_pool);
    OS_ASSERT(msg_pool_size > 0U);
    OS_ASSERT(msg_size > 0U);
    OS_ASSERT(os_is_irq_active() == OS_FALSE);

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue);

    os_spin_lock(&gs_os_mq_resource_list_lock);
    os_list_for_each(pos, &gs_os_mq_resource_list_head)
    {
        iter_mq = os_list_entry(pos, os_msgqueue_t, resource_node);
        if (iter_mq == mq)
        {
            OS_KERN_LOG(KERN_ERROR, MQ_TAG, "The mq(addr: %p, name: %s) has been exist", iter_mq, iter_mq->name);

            mq = OS_NULL;
            break;
        }
    }

    if (OS_NULL != mq)
    {
        os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
        os_spin_unlock(&gs_os_mq_resource_list_lock);

        ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, msg_size, OS_ALLOC_TYPE_STATIC);
        if (OS_SUCCESS == ret)
        {
            mq->object_inited = OS_KOBJ_INITED;
        }
        else
        {
            os_spin_lock(&gs_os_mq_resource_list_lock);
            os_list_del(&mq->resource_node);
            os_spin_unlock(&gs_os_mq_resource_list_lock);
        }
    }
    else
    {
        os_spin_unlock(&gs_os_mq_resource_list_lock);
    }

    return OS_TYPE_CONVERT(os_msgqueue_id, mq);
}

分步详解：

第一步：参数校验（条件：控制块/内存池非空、大小非 0、非中断上下文）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != msgqueue);
    OS_ASSERT(OS_NULL != msg_pool);
    OS_ASSERT(msg_pool_size > 0U);
    OS_ASSERT(msg_size > 0U);
    OS_ASSERT(os_is_irq_active() == OS_FALSE);

进入本步的条件：调用 os_msgqueue_create_static() 时无条件执行（本步是入口硬性校验，不满足条件直接断言崩溃，不返回错误码）。

详细解释：

• OS_ASSERT(OS_NULL != msgqueue)：条件 msgqueue != OS_NULL——控制块必须非空（用户用 OS_MSGQUEUE_DEFINE 定义的变量地址）。
• OS_ASSERT(OS_NULL != msg_pool)：条件 msg_pool != OS_NULL——内存池必须非空（用户用 OS_MSGQUEUE_POOL_DEFINE 定义的数组）。
• OS_ASSERT(msg_pool_size > 0U)：条件 msg_pool_size > 0——内存池大小必须非 0。
• OS_ASSERT(msg_size > 0U)：条件 msg_size > 0——消息大小必须非 0。
• OS_ASSERT(os_is_irq_active() == OS_FALSE)：条件 os_is_irq_active() == OS_FALSE——不能在中断上下文创建。

关键点：与动态创建不同，静态创建不分配内存（控制块、内存池都由用户提供），所以本步校验的是用户传入的指针/大小合法性。

第二步：防重复创建检查（条件：第一步断言通过）—— 遍历资源链表，已存在则置 mq = OS_NULL

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue);

    os_spin_lock(&gs_os_mq_resource_list_lock);
    os_list_for_each(pos, &gs_os_mq_resource_list_head)
    {
        iter_mq = os_list_entry(pos, os_msgqueue_t, resource_node);
        if (iter_mq == mq)
        {
            OS_KERN_LOG(KERN_ERROR, MQ_TAG, "The mq(addr: %p, name: %s) has been exist", iter_mq, iter_mq->name);

            mq = OS_NULL;
            break;
        }
    }

进入本步的条件：第一步五条断言全部通过（参数合法），函数继续执行。

详细解释：

• mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue)：把用户传入的 dummy 控制块地址强转为真实内核控制块类型。
• 加资源链表锁 gs_os_mq_resource_list_lock（保护共享链表，防止并发创建竞争）。
• os_list_for_each(pos, &gs_os_mq_resource_list_head)：遍历整个消息队列资源链表，逐个取出 iter_mq 比较。
• 子分支：条件 iter_mq == mq（发现同一个控制块地址已在链表中，即重复创建）：
  • 打印错误日志 has been exist；
  • 置 mq = OS_NULL 并 break——用"置空"来标记"此队列已存在、不可再创建"。

关键点：静态创建需要防重复，是因为静态控制块内存由用户提供、地址固定。如果同一个 msgqueue 地址被创建两次，会重复登记进资源链表、重复初始化同一块内存，造成链表错乱。而动态创建每次 malloc 新地址，天然不重复，故无此检查。

第三步：登记并初始化（条件：mq != OS_NULL，即未重复创建）—— 登记链表 + 初始化 + 标记已初始化

    if (OS_NULL != mq)
    {
        os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node);
        os_spin_unlock(&gs_os_mq_resource_list_lock);

        ret = _k_mq_init(mq, name, msg_pool, msg_pool_size, msg_size, OS_ALLOC_TYPE_STATIC);
        if (OS_SUCCESS == ret)
        {
            mq->object_inited = OS_KOBJ_INITED;
        }
        else
        {
            os_list_del(&mq->resource_node);   /* 初始化失败，从链表移除 */
        }
    }
    else
    {
        os_spin_unlock(&gs_os_mq_resource_list_lock);
    }

进入本步的条件：第二步未发现重复，即 mq != OS_NULL（else 分支是重复时仅释放锁）。

详细解释：

• os_list_add_tail(&gs_os_mq_resource_list_head, &mq->resource_node)：把队列登记进消息队列资源链表（此时仍持锁）。
• os_spin_unlock(...)：登记完释放锁（后续初始化无需再持锁）。
• ret = _k_mq_init(..., OS_ALLOC_TYPE_STATIC)：初始化控制块（注意分配类型是 OS_ALLOC_TYPE_STATIC，与动态的 OS_KOBJ_ALLOC_TYPE_DYNAMIC 区分——销毁时据此决定是否释放内存）。
• 子分支 A：条件 ret == OS_SUCCESS（初始化成功）：mq->object_inited = OS_KOBJ_INITED 标记已初始化。
• 子分支 B：条件 ret != OS_SUCCESS（初始化失败）：os_list_del(&mq->resource_node) 从资源链表移除该队列（回滚，避免留下一个半初始化的链节点）。

最终返回值：return OS_TYPE_CONVERT(os_msgqueue_id, mq)——成功返回队列 ID；重复创建时 mq == OS_NULL，故返回 OS_NULL。

2.4 销毁消息队列

os_err_t os_msgqueue_destroy(os_msgqueue_id msgqueue_id);

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_destroy(os_msgqueue_id msgqueue_id)
{
    os_bool_t      need_sched;
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_FALSE == os_is_irq_active());
#ifndef OS_USING_HEAP
    OS_ASSERT(OS_ALLOC_TYPE_STATIC == mq->object_alloc_type);
#endif

    need_sched = OS_FALSE;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    mq->object_inited = OS_KOBJ_DEINITED;

    if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
    {
        need_sched = OS_TRUE;
    }
    if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
    {
        need_sched = OS_TRUE;
    }

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_unlock_irq(&mq->lock, irq_save);
    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_ENTER();
        OS_KERNEL_EXIT_SCHED();
    }
#else
    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_EXIT_SCHED();
    }
    else
    {
        OS_KERNEL_EXIT();
    }
#endif

    os_spin_lock(&gs_os_mq_resource_list_lock);
    os_list_del(&mq->resource_node);
    os_spin_unlock(&gs_os_mq_resource_list_lock);

    if (OS_KOBJ_ALLOC_TYPE_DYNAMIC == mq->object_alloc_type)
    {
        OS_KERNEL_FREE(mq->msg_pool);
        mq->msg_pool = OS_NULL;

        OS_KERNEL_FREE(mq);
        mq = OS_NULL;
    }

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、非中断上下文、非堆模式下必须静态分配）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_FALSE == os_is_irq_active());
#ifndef OS_USING_HEAP
    OS_ASSERT(OS_ALLOC_TYPE_STATIC == mq->object_alloc_type);
#endif

进入本步的条件：调用 os_msgqueue_destroy() 时无条件执行（本步是入口硬性校验，不满足条件直接断言崩溃，不返回错误码）。

详细解释：

• OS_ASSERT(OS_NULL != mq)：条件 mq != OS_NULL——要销毁的队列必须非空。
• OS_ASSERT(OS_KOBJ_INITED == mq->object_inited)：条件 object_inited == OS_KOBJ_INITED——队列必须是"已初始化"状态，防重复销毁（销毁后 object_inited 会变 DEINITED，再次销毁会在这里断言崩溃）。
• OS_ASSERT(OS_FALSE == os_is_irq_active())：条件 os_is_irq_active() == OS_FALSE——不能在中断中销毁（销毁可能触发调度、释放内存）。
• OS_ASSERT(OS_ALLOC_TYPE_STATIC == mq->object_alloc_type)：条件（仅当未启用堆 OS_USING_HEAP 时编译）object_alloc_type == OS_ALLOC_TYPE_STATIC——未启用堆时只能销毁静态创建的队列（因为动态创建需要 OS_KERNEL_FREE，无堆则无法释放）。

关键点：销毁前必须先确认队列是"已初始化"状态，且类型与分配模式匹配，否则断言崩溃。

第二步：标记去初始化并唤醒所有阻塞任务（条件：第一步断言通过）—— 标记 DEINITED + 唤醒所有阻塞的收发任务

    mq->object_inited = OS_KOBJ_DEINITED;

    if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
    {
        need_sched = OS_TRUE;
    }
    if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
    {
        need_sched = OS_TRUE;
    }

进入本步的条件：第一步四条断言全部通过（队列合法、已初始化），函数继续执行。

详细解释：

• mq->object_inited = OS_KOBJ_DEINITED：先把队列标记为"去初始化"——防止销毁过程中其他任务（如 send/recv）误操作这个正在销毁的队列。
• if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))：条件 send_task_list_head 上有阻塞任务时，唤醒所有因队列满而阻塞的发送者，返回 OS_TRUE 则置 need_sched = OS_TRUE。
• if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))：条件 recv_task_list_head 上有阻塞任务时，唤醒所有因队列空而阻塞的接收者，返回 OS_TRUE 则置 need_sched = OS_TRUE。

关键点：销毁队列前必须唤醒所有仍阻塞的收发任务，否则这些任务会永远挂在一个已销毁队列的链表上，造成任务永久挂死。唤醒的动作由 k_cancel_all_blocked_task 完成。

第三步：触发调度 + 移除资源链表 + 释放内存（条件：取决于 need_sched 和 object_alloc_type）—— 收尾清理

    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_EXIT_SCHED();
    }
    else
    {
        OS_KERNEL_EXIT();
    }

    os_spin_lock(&gs_os_mq_resource_list_lock);
    os_list_del(&mq->resource_node);
    os_spin_unlock(&gs_os_mq_resource_list_lock);

    if (OS_KOBJ_ALLOC_TYPE_DYNAMIC == mq->object_alloc_type)
    {
        OS_KERNEL_FREE(mq->msg_pool);   /* 释放内存池 */
        mq->msg_pool = OS_NULL;

        OS_KERNEL_FREE(mq);             /* 释放控制块 */
        mq = OS_NULL;
    }

    return OS_SUCCESS;

进入本步的条件：第二步已执行完，本步为收尾清理，按两个子条件分别处理。

详细解释：

• 子条件 A：need_sched == OS_TRUE（第二步唤醒了阻塞任务）→ OS_KERNEL_EXIT_SCHED() 触发一次调度，让被唤醒的任务有机会运行；否则（need_sched == OS_FALSE，无唤醒）→ 仅 OS_KERNEL_EXIT() 退出临界区。
• os_list_del(&mq->resource_node)：加资源链表锁，把该队列的 resource_node 从消息队列资源链表移除（销毁后全局不可再查询到它）。
• 子条件 B：mq->object_alloc_type == OS_KOBJ_ALLOC_TYPE_DYNAMIC（该队列是动态创建的）→ 释放 mq->msg_pool（内存池）和 mq（控制块），并置空；静态创建的队列不释放（控制块、内存池都由用户静态提供，不归内核管理，释放会造成非法 free）。
• 返回 OS_SUCCESS（销毁必然成功，因为所有失败路径都在断言处崩溃）。

关键点：动态 vs 静态创建的销毁差异——只有动态创建的才 OS_KERNEL_FREE；静态创建的不能 free 用户提供的内存。这一步也解释了为什么创建时要在 object_alloc_type 里区分两种类型。

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 队列满

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_send(os_msgqueue_id msgqueue_id, const void *buffer, os_size_t buff_size, os_tick_t timeout)
{
    os_err_t                ret;
    os_task_t              *block_task;
    os_bool_t               need_sched;
    register os_msgqueue_t *mq;
    register os_mq_msg_t   *msg;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_KERNEL_INIT();

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

    ret        = OS_SUCCESS;
    need_sched = OS_FALSE;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    if (OS_NULL != mq->msg_queue_free)
    {
        _k_mq_get_free_msg(mq, msg);
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_unlock_irq(&mq->lock, irq_save);
#else
        OS_KERNEL_EXIT();
#endif
#endif
        (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
        msg->msg_len = buff_size;
    }
    else
    {
        if (OS_NO_WAIT == timeout)
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
            ret = OS_FULL;
        }
        else
        {
            OS_ASSERT((OS_NULL != _k_task_self()));
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
            }

            if (OS_SUCCESS == ret)
            {
                /*
                 * swap_data field filled by the receiver is point to free message buffer,
                 * so put the sending message into this buffer.
                 */
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;
                (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
                msg->msg_len = buff_size;

#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
                os_spin_lock_irq(&mq->lock, &irq_save);
                k_kernel_spin_lock();
#else
                OS_KERNEL_ENTER();
#endif
#endif
            }
        }
    }

    if (OS_SUCCESS == ret)
    {
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_lock_irq(&mq->lock, &irq_save);
        k_kernel_spin_lock();
#else
        OS_KERNEL_ENTER();
#endif
#endif
        if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
        {
            /*
             * When the receiver task has been blocked, put address of the message buffer filled data on swap_data,
             * and unblock the receiver task. The receiver task will get message from swap_data.
             */
            block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
            need_sched = k_unblock_task(block_task);

            block_task->swap_data = (os_ubase_t)msg;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
            os_spin_unlock_irq(&mq->lock, irq_save);

            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_ENTER();
                OS_KERNEL_EXIT_SCHED();
            }
#else
            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_EXIT_SCHED();
            }
            else
            {
                OS_KERNEL_EXIT();
            }
#endif
        }
        else
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
#endif

            _k_mq_put_msg_to_queue(mq, msg);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
        }
    }

    return ret;
}

分步详解：

第一步：准备与参数校验（条件：队列/缓冲区非空、buff_size 合法、已初始化、等待条件合法）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

进入本步的条件：调用 os_msgqueue_send() 时无条件执行（本步是入口硬性校验，不满足条件直接断言崩溃，不返回错误码）。

详细解释：

断言	作用
mq != OS_NULL	消息队列非空
buffer != OS_NULL	发送缓冲区非空
(buff_size > 0) && (buff_size <= max_msg_size)	消息大小必须 > 0 且 ≤ 队列单条消息上限
object_inited == OS_KOBJ_INITED	消息队列必须已初始化
多条 timeout 断言	需要真正阻塞等待时，不能在中断/关中断/调度锁定中调用

第二步：队列有空闲槽位（条件 msg_queue_free != OS_NULL）—— 直接放入

进入本步的条件：最外层 if (OS_NULL != mq->msg_queue_free) 为真，即队列中还有空闲消息块（队列未满）。满足此条件时走 if 分支，直接放入消息。

    if (OS_NULL != mq->msg_queue_free)
    {
        _k_mq_get_free_msg(mq, msg);
        (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
        msg->msg_len = buff_size;
    }

详细解释：

• 条件 msg_queue_free != OS_NULL：队列有空闲消息槽位（msg_queue_free 指向第一个空闲消息块，非空表示还有空位）。
• _k_mq_get_free_msg(mq, msg)：取一个空闲消息块，msg 指向该消息块的起始地址。
• memcpy(...)：把数据拷贝到消息块的消息体（跳过消息头 sizeof(os_mq_msg_hdr_t)），完成"数据拷贝"（详细语法见下方）。
• msg->msg_len = buff_size：记录消息长度。

注意：本步走的是外层 if 的真分支；若条件为假（队列满），则落入 第三步 的 else 分支。

memcpy(...) 语法详细拆解：

这句的完整代码是：

(void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);

它调用 C 标准库函数 memcpy(目标地址, 源地址, 拷贝字节数)，把 buffer 指向的数据拷贝到消息块的消息体。逐个部分拆解：

部分	代码	含义
目标地址	(void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t))	消息块中消息体的起始地址（跳过消息头）
源地址	buffer	用户要发送的数据缓冲区
拷贝字节数	buff_size	要拷贝的数据大小
返回值强转	(void)memcpy(...)	忽略 memcpy 的返回值（memcpy 返回目标地址）

目标地址 (uint8_t *)msg + sizeof(os_mq_msg_hdr_t) 的指针运算详解：

这一步是核心，它用指针算术计算"消息体的地址"。消息块的内存布局是：前部分是消息头，后部分是消息体：

消息块内存布局（msg 指向起始地址）：
┌────────────────────────────────────────┐
│ 消息头 os_mq_msg_hdr_t                │  ← 偏移 0，sizeof(os_mq_msg_hdr_t) 字节
├────────────────────────────────────────┤
│ 消息体（存放实际数据）                  │  ← 偏移 sizeof(os_mq_msg_hdr_t) 开始
└────────────────────────────────────────┘
  地址：msg ────────────────────────► msg + sizeof(os_mq_msg_hdr_t)

指针运算过程：

步骤	代码	说明
①	(uint8_t *)msg	把 msg（os_mq_msg_t * 类型）强转为 uint8_t *（字节指针）
②	+ sizeof(os_mq_msg_hdr_t)	字节指针加偏移量，跳过消息头，指向消息体起始
③	(void *)	把结果转为 void *（memcpy 第一参数要求 void *）

为什么必须先转成 uint8_t *（字节指针）再偏移？

这是指针算术的关键规则：指针加整数，移动的字节数 = 整数 × 指针所指类型的大小。

指针类型	ptr + 1 移动多少字节
uint8_t *（1 字节）	1 字节
int *（4 字节）	4 字节
os_mq_msg_t *（消息块结构）	整个消息块的大小

如果我们直接用 msg + 1（不转 uint8_t *），os_mq_msg_t * 会让它移动一个完整消息块的大小，而不是跳过消息头。所以必须先转成 uint8_t *，让 + sizeof(...) 精确地按字节偏移，从而恰好跳过消息头、指向消息体。

一句话：(uint8_t *)msg + sizeof(os_mq_msg_hdr_t) 就是"把指针当作字节地址，向后偏移消息头的大小，得到消息体起始地址"，然后 memcpy 把数据写进去。由于是先强转成字节指针再偏移，偏移量精确到字节，不会错位。

第三步：队列满（条件 msg_queue_free == OS_NULL）—— 等待或返回满

进入本步的条件：最外层 if (OS_NULL != mq->msg_queue_free) 为假，即队列已满（msg_queue_free == OS_NULL），走 else 分支。此时本步内部又分两个子分支，由 timeout 决定。

    else
    {
        if (OS_NO_WAIT == timeout)
        {
            ret = OS_FULL;   /* 不等待，返回队列满 */
        }
        else
        {
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
            }

            if (OS_SUCCESS == ret)
            {
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;   /* 接收者预留的空闲消息块 */
                (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
                msg->msg_len = buff_size;
            }
        }
    }

详细解释（本步内部按 timeout 分两个子分支）：

子分支 A：条件 timeout == OS_NO_WAIT（不等待）：

• 队列已满且调用方不想等待，直接返回 OS_FULL（队列已满，发送失败）。

子分支 B：条件 timeout != OS_NO_WAIT（愿意等待），内部再按唤醒策略分两步：

• 条件 OS_MQ_WAKE_TYPE_PRIO == mq->wake_type：按优先级唤醒，k_block_task(..., OS_TRUE)。
• 否则（FIFO）：k_block_task(..., OS_FALSE)，按等待时间唤醒。
• k_block_task() 把当前任务挂到发送者等待队列（send_task_list_head）并让出 CPU，直到接收者取走消息腾出空位、唤醒它。
• 被唤醒后，条件 ret == OS_SUCCESS（成功被唤醒）：从 swap_data 拿到接收者预留的空闲消息块，拷贝数据并记录长度。
• 若 ret != OS_SUCCESS（超时等其他原因）：ret 保持错误值，不拷贝数据。

"从 swap_data 拿到接收者预留的空闲消息块"到底是怎么发生的？

这句是整个机制中最精巧的部分，需要结合发送者阻塞和接收者取消息两端来理解。它的核心是一个"消息块直接转交（接力）"机制，避免"释放消息块再重新分配"的往返开销。

背景：发送者为什么会阻塞在 swap_data 上？

当队列满时，发送者调用 k_block_task() 把自己挂到发送者等待队列，然后让出 CPU。此时发送者"停"在 k_block_task() 这一行，等待被唤醒。

发送者（队列满，阻塞中）：
    ret = k_block_task(...);          ← 发送者停在这里，等待被唤醒
    if (OS_SUCCESS == ret)            ← 被唤醒后从这里继续
    {
        msg = (os_mq_msg_t *)_k_task_self()->swap_data;   ← 关键：从这里拿消息块
        ...
    }

接收者如何"腾出空位并转交消息块"？

当接收者调用 os_msgqueue_recv() 取走一条消息后，如果发现有发送者阻塞在等待，它不会把取出的消息块简单释放回空闲池，而是把这个消息块的地址直接交给发送者（见 recv 分步详解）：

/* 接收者 os_msgqueue_recv() 中：有发送者阻塞时 */
block_task            = os_list_first_entry(&mq->send_task_list_head, os_task_t, task_node);  /* 取阻塞的发送者 */
need_sched            = k_unblock_task(block_task);      /* 唤醒发送者 */
block_task->swap_data = (os_ubase_t)msg;                  /* 把这个空闲消息块地址塞进发送者的 swap_data */

两端如何通过 swap_data 接力？

            发送者（阻塞）                        接收者
                │                                   │
① 队列满，发送者 k_block_task 阻塞  ←────────────   │
                │                                   │
② 接收者 recv 取走一条消息，腾出空位                │
                │                       ┌───────────┘
                │                       ▼
③ 接收者：把腾出的消息块地址写入 发送者.swap_data
            并 k_unblock_task 唤醒发送者
                │
                ▼
④ 发送者被唤醒，从 _k_task_self()->swap_data
   拿到消息块地址 msg
                │
                ▼
⑤ 发送者 memcpy 数据到该消息块，msg->msg_len = 长度
   （消息块已就绪，直接可用）

为什么能这样接力？关键在"消息块已入队又出队"：

• 接收者取走的是一条已经入队的消息块（它刚被释放出队）。
• 正常情况下，接收者应把这个消息块 _k_mq_release_free_msg() 释放回空闲池。
• 但发送者正等着发送新的数据，需要一块空闲消息块。
• 所以接收者直接把这块刚腾出的消息块交给发送者，发送者唤醒后立即复用它，无需"释放回池再取"。

这句 msg = (os_mq_msg_t *)_k_task_self()->swap_data; 的含义：

msg = (os_mq_msg_t *)_k_task_self()->swap_data;

部分	含义
_k_task_self()	获取当前任务（即被唤醒的发送者自己）
->swap_data	读取该任务的 swap_data 字段（接收者写入的消息块地址）
(os_mq_msg_t *)	把地址强转为消息块指针，赋给 msg
msg->msg_len = buff_size	接着拷贝数据并记录长度

一句话总结：发送者因队列满而阻塞时，接收者取走消息后把腾出的空闲消息块地址通过 swap_data 直接转交给发送者；发送者唤醒后从自己的 swap_data 拿到这个现成的消息块，直接拷贝数据——省去了"消息块释放回池再重新获取"的往返，是消息队列内存池高效复用的关键设计。

第四步：消息就绪后 —— 唤醒接收者（条件 recv_task_list_head 非空）或放入队列

进入本步的条件：外层 if (OS_SUCCESS == ret) 为真，即前面的发送已成功（无论是直接放入还是阻塞唤醒后拿到消息块，msg 都已填充好数据）。本步内部再按"是否有接收者阻塞"分两个子分支。

    if (OS_SUCCESS == ret)
    {
        if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
        {
            /*
             * When the receiver task has been blocked, put address of the message buffer filled data on swap_data,
             * and unblock the receiver task. The receiver task will get message from swap_data.
             */
            block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
            need_sched = k_unblock_task(block_task);

            block_task->swap_data = (os_ubase_t)msg;

            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_EXIT_SCHED();
            }
            else
            {
                OS_KERNEL_EXIT();
            }
        }
        else
        {
            _k_mq_put_msg_to_queue(mq, msg);

            OS_KERNEL_EXIT();
        }
    }

详细解释（本步内部按"是否有接收者阻塞"分两个子分支）：

子分支 A：条件 recv_task_list_head 非空（有接收者阻塞）：

• 说明有接收者正因队列空而阻塞等待消息。
• 此时不把消息入队，而是直接把消息块地址放入接收者的 swap_data，k_unblock_task() 唤醒它（接收者直接从 swap_data 取消息）。
• 内部再按条件 need_sched == OS_TRUE（唤醒需要调度）：触发调度 OS_KERNEL_EXIT_SCHED()；否则仅退出临界区 OS_KERNEL_EXIT()。

子分支 B：条件 recv_task_list_head 为空（无接收者阻塞）：

• 没有接收者等待，用 _k_mq_put_msg_to_queue() 把消息放入队列（队尾），等待后续接收者取走。
• 退出临界区。

关键点：发送时如果已有接收者在等，消息直接转交给接收者（通过 swap_data），不入队；否则才放入队列。这避免了"入队再立即出队"的冗余操作。

os_msgqueue_send 的 if-else 条件走向总结（对照源码 823-936 行）：

开始
└─ OS_ASSERT 参数校验（mq / buffer / buff_size / 初始化 / 等待条件）任一失败 → 直接断言崩溃（不返回）
进入临界区后判断：if (msg_queue_free != OS_NULL)（队列有空闲槽位？第二步）
   ├─ 是 ──► _k_mq_get_free_msg() 取空闲消息块 + memcpy 拷贝数据 → 返回 OS_SUCCESS
   └─ 否 ──► else（队列满，第三步）
             ├─ if (timeout == OS_NO_WAIT)（不等待）→ 释放临界区，ret = OS_FULL
             └─ else（timeout != OS_NO_WAIT，等待）
                   ├─ if (wake_type == OS_MQ_WAKE_TYPE_PRIO) → k_block_task(..., OS_TRUE)
                   └─ else（FIFO）→ k_block_task(..., OS_FALSE)
                   └─ 被唤醒后：if (ret == OS_SUCCESS) → 从 self->swap_data 取消息块 + memcpy
统一出口：if (ret == OS_SUCCESS)（第四步，发送成功才进入）
   ├─ if (&recv_task_list_head 非空，有接收者在等)（第五步）
   │     ├─ 取出接收者 block_task + k_unblock_task() + 把 msg 地址写入其 swap_data
   │     └─ if (need_sched == OS_TRUE) → 触发调度（SMP 下同样）
   └─ else（无接收者在等）→ _k_mq_put_msg_to_queue() 入队
最终 return ret

使用示例：

os_err_t ret = os_msgqueue_send(mq_id, &data, sizeof(data), OS_WAIT_FOREVER);
if (ret != OS_EOK)
{
    os_kprintf("send failed: %d\r\n", ret);
}

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);

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

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_send_urgent(os_msgqueue_id msgqueue_id, const void *buffer, os_size_t buff_size, os_tick_t timeout)
{
    os_err_t                ret;
    os_task_t              *block_task;
    os_bool_t               need_sched;
    register os_msgqueue_t *mq;
    register os_mq_msg_t   *msg;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_KERNEL_INIT();

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

    ret        = OS_SUCCESS;
    need_sched = OS_FALSE;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    if (OS_NULL != mq->msg_queue_free)
    {
        _k_mq_get_free_msg(mq, msg);
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_unlock_irq(&mq->lock, irq_save);
#else
        OS_KERNEL_EXIT();
#endif
#endif
        (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
        msg->msg_len = buff_size;
    }
    else
    {
        if (OS_NO_WAIT == timeout)
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
            ret = OS_FULL;
        }
        else
        {
            OS_ASSERT((OS_NULL != _k_task_self()));
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
            }

            if (OS_SUCCESS == ret)
            {
                /*
                 * swap_data field filled by the receiver is point to free message buffer,
                 * so put the sending message into this buffer.
                 */
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;
                (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
                msg->msg_len = buff_size;

#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
                os_spin_lock_irq(&mq->lock, &irq_save);
                k_kernel_spin_lock();
#else
                OS_KERNEL_ENTER();
#endif
#endif
            }
        }
    }

    if (OS_SUCCESS == ret)
    {
#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_lock_irq(&mq->lock, &irq_save);
        k_kernel_spin_lock();
#else
        OS_KERNEL_ENTER();
#endif
#endif
        if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
        {
            /*
             * When the receiver task has been blocked, put address of the message buffer filled data on swap_data,
             * and unblock the receiver task. The receiver task will get message from swap_data.
             */
            block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
            need_sched = k_unblock_task(block_task);

            block_task->swap_data = (os_ubase_t)msg;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
            os_spin_unlock_irq(&mq->lock, irq_save);

            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_ENTER();
                OS_KERNEL_EXIT_SCHED();
            }
#else
            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_EXIT_SCHED();
            }
            else
            {
                OS_KERNEL_EXIT();
            }
#endif
        }
        else
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
#endif

            _k_mq_put_urgentmsg_to_queue(mq, msg);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
        }
    }

    return ret;
}

分步详解：

os_msgqueue_send_urgent() 与 os_msgqueue_send() 的流程几乎完全一致（参数校验、取空闲块、队列满时等待、唤醒接收者都相同），唯一区别在最后一步无接收者时插入队列的位置。

第一步：准备与参数校验（条件：队列/缓冲区非空、buff_size 合法、已初始化、等待条件合法）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT((buff_size > 0U) && (buff_size <= mq->max_msg_size));
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

进入本步的条件：调用 os_msgqueue_send_urgent() 时无条件执行（与 send 完全相同的八条硬性断言校验）。

详细解释（与 send 的第一步逐条一致）：

• mq != OS_NULL：队列必须非空；buffer != OS_NULL：发送缓冲区非空。
• buff_size > 0U && buff_size <= mq->max_msg_size：消息长度必须合法且不超过队列单条上限。
• object_inited == OS_KOBJ_INITED：队列已初始化。
• 三条"等待条件"断言：需要阻塞等待（timeout != OS_NO_WAIT）时，必须有当前任务、不在中断、未关中断、未锁调度。
• timeout < OS_TICK_MAX/2 || timeout == OS_WAIT_FOREVER：超时值合法。

第二步：队列有空闲槽位（条件 msg_queue_free != OS_NULL）—— 直接取块拷贝数据

    if (OS_NULL != mq->msg_queue_free)
    {
        _k_mq_get_free_msg(mq, msg);
        (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
        msg->msg_len = buff_size;
    }

进入本步的条件：msg_queue_free != OS_NULL（消息池里还有空闲消息块，队列未满）。流程与 send 的第二步完全相同——取空闲块、memcpy 数据、记录长度。

第三步：队列满（条件 msg_queue_free == OS_NULL）—— 等待或返回满

    else
    {
        if (OS_NO_WAIT == timeout)
        {
            ret = OS_FULL;
        }
        else
        {
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->send_task_list_head, timeout, OS_FALSE);
            }
            if (OS_SUCCESS == ret)
            {
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;
                (void)memcpy((void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), buffer, buff_size);
                msg->msg_len = buff_size;
            }
        }
    }

进入本步的条件：msg_queue_free == OS_NULL（队列满，无空闲消息块）。

• 子分支 A：条件 timeout == OS_NO_WAIT（不等待）→ 直接 ret = OS_FULL。
• 子分支 B：条件 timeout != OS_NO_WAIT（等待）→ 按 wake_type 阻塞挂到 send_task_list_head；被唤醒且 ret == OS_SUCCESS 时从 swap_data 取消息块拷贝数据。与 send 完全相同。

第四步：消息就绪后（条件 ret == OS_SUCCESS）—— 唤醒接收者（条件 recv_task_list_head 非空）或插入队首

    if (OS_SUCCESS == ret)
    {
        if (&mq->recv_task_list_head != mq->recv_task_list_head.next)
        {
            block_task = os_list_first_entry(&mq->recv_task_list_head, os_task_t, task_node);
            need_sched = k_unblock_task(block_task);
            block_task->swap_data = (os_ubase_t)msg;
            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_EXIT_SCHED();
            }
            else
            {
                OS_KERNEL_EXIT();
            }
        }
        else
        {
            _k_mq_put_urgentmsg_to_queue(mq, msg);   /* ★ 与 send 唯一区别：插入队首 */
            OS_KERNEL_EXIT();
        }
    }
    return ret;

进入本步的条件：ret == OS_SUCCESS（前面取块/等待成功，消息已填充好数据）。

• 子分支 A：条件 recv_task_list_head 非空（有接收者阻塞）→ 与 send 完全相同：取队首接收者、写 swap_data、k_unblock_task 唤醒、need_sched 时触发调度。
• 子分支 B：条件 recv_task_list_head 为空（无接收者）→ 这里是 send_urgent 与 send 的唯一区别：调用 _k_mq_put_urgentmsg_to_queue(mq, msg) 把消息插入队首（而 send 用 _k_mq_put_msg_to_queue 插队尾）。

与 send 的差异对比表：

对比项	普通发送 send	紧急发送 send_urgent
第一步~第三步	完全相同	完全相同
有接收者等待	直接转交给接收者（swap_data）	直接转交给接收者（swap_data）——相同
无接收者时入队位置	_k_mq_put_msg_to_queue() 插入队尾	_k_mq_put_urgentmsg_to_queue() 插入队首
出队顺序	先进先出（FIFO）	紧急消息优先出队

关键点：唯一区别是 _k_mq_put_urgentmsg_to_queue()（插队首）vs _k_mq_put_msg_to_queue()（插队尾）。所以紧急消息优先被接收者取出，适合处理高优先级/紧急事件（如告警、超时信号）。

使用示例：

const char *urgent = "ALARM!";
os_msgqueue_send_urgent(mq_id, urgent, strlen(urgent) + 1, OS_NO_WAIT);   /* 插入队首，优先出队 */

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 队列空

完整源码（kernel/source/os_mq.c，与项目一致）：

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)
{
    os_task_t              *block_task;
    os_err_t                ret;
    os_bool_t               need_sched;
    register os_msgqueue_t *mq;
    register os_mq_msg_t   *msg;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT(buff_size > 0U);
    OS_ASSERT(OS_NULL != recv_msg_size);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

    ret = OS_SUCCESS;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    if (OS_NULL == mq->msg_queue_head)
    {
        if (OS_NO_WAIT == timeout)
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
            ret = OS_EMPTY;
        }
        else
        {
            OS_ASSERT((OS_NULL != _k_task_self()));
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_FALSE);
            }

            if (OS_SUCCESS == ret)
            {
#ifdef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
                os_spin_lock_irq(&mq->lock, &irq_save);
                k_kernel_spin_lock();
#else
                OS_KERNEL_ENTER();
#endif
#endif
                /*
                 * When the sender wakes up this task, it puts address of the message buffer on the swap_data,
                 * so there's no need to get message from the queue.
                 */
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;
            }
        }
    }
    else
    {
        _k_mq_get_msg_from_queue(mq, msg);

#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_unlock_irq(&mq->lock, irq_save);
#else
        OS_KERNEL_EXIT();
#endif
#endif
    }

    if (OS_SUCCESS == ret)
    {
        if (buff_size < msg->msg_len)
        {
            /* When receive buffer size is less than message size, discard this message. */
            OS_KERN_LOG(KERN_ERROR,
                        MQ_TAG,
                        "Recv buff size(%lu) of task(%s) is less than msg size(%lu)",
                        buff_size,
                        _k_task_self()->name,
                        msg->msg_len);
            ret = OS_NOMEM;
        }
        else
        {
            (void)memcpy(buffer, (void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), msg->msg_len);
            *recv_msg_size = msg->msg_len;
        }

#ifndef MQ_FULL_LOCK
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
        os_spin_lock_irq(&mq->lock, &irq_save);
        k_kernel_spin_lock();
#else
        OS_KERNEL_ENTER();
#endif
#endif
        if (&mq->send_task_list_head != mq->send_task_list_head.next)
        {
            /*
             * When the sender task was blocked, the message buffer that has just been processed will
             * not be released into free pool. Instead, this message buffer is assigned to the sender task.
             */
            block_task            = os_list_first_entry(&mq->send_task_list_head, os_task_t, task_node);
            need_sched            = k_unblock_task(block_task);
            block_task->swap_data = (os_ubase_t)msg;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
            os_spin_unlock_irq(&mq->lock, irq_save);

            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_ENTER();
                OS_KERNEL_EXIT_SCHED();
            }
#else
            if (OS_TRUE == need_sched)
            {
                OS_KERNEL_EXIT_SCHED();
            }
            else
            {
                OS_KERNEL_EXIT();
            }
#endif
        }
        else
        {
#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            k_kernel_spin_unlock();
#endif

            _k_mq_release_free_msg(mq, msg);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
            os_spin_unlock_irq(&mq->lock, irq_save);
#else
            OS_KERNEL_EXIT();
#endif
        }
    }

    return ret;
}

分步详解：

第一步：准备与参数校验（条件：队列/缓冲区/传出参数非空、buff_size > 0、已初始化、等待条件合法）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_NULL != buffer);
    OS_ASSERT(buff_size > 0U);
    OS_ASSERT(OS_NULL != recv_msg_size);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_active()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_irq_disabled()));
    OS_ASSERT((OS_NO_WAIT == timeout) || (OS_NULL == k_task_self()) || (OS_FALSE == os_is_schedule_locked()));
    OS_ASSERT((timeout < (OS_TICK_MAX / 2)) || (OS_WAIT_FOREVER == timeout));

进入本步的条件：调用 os_msgqueue_recv() 时无条件执行（本步是入口硬性校验，不满足条件直接断言崩溃，不返回错误码）。

详细解释：

• OS_ASSERT(OS_NULL != mq)：条件 mq != OS_NULL——消息队列必须非空。
• OS_ASSERT(OS_NULL != buffer)：条件 buffer != OS_NULL——接收缓冲区必须非空（否则拷贝无处可放）。
• OS_ASSERT(buff_size > 0U)：条件 buff_size > 0——接收缓冲区大小必须非 0。
• OS_ASSERT(OS_NULL != recv_msg_size)：条件 recv_msg_size != OS_NULL——传出参数必须非空（函数要往它写入实际接收大小，若为 NULL 会解引用崩溃）。
• OS_ASSERT(OS_KOBJ_INITED == mq->object_inited)：条件 object_inited == OS_KOBJ_INITED——队列已初始化。
• 三条"等待条件"断言：需要阻塞等待（timeout != OS_NO_WAIT）时，必须有当前任务、不在中断、未关中断、未锁调度。
• OS_ASSERT((timeout < OS_TICK_MAX/2) || (timeout == OS_WAIT_FOREVER))：条件超时值合法。

关键点：注意 recv_msg_size 是传出参数（函数要写它），判空尤为重要——这是与 send 断言不同的地方。

第二步：队列为空（条件 msg_queue_head == OS_NULL）—— 等待或返回空

    if (OS_NULL == mq->msg_queue_head)
    {
        if (OS_NO_WAIT == timeout)
        {
            ret = OS_EMPTY;   /* 不等待，队列空返回 */
        }
        else
        {
            if (OS_MQ_WAKE_TYPE_PRIO == mq->wake_type)
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_TRUE);
            }
            else
            {
                ret = k_block_task(&mq->lock, irq_save, &mq->recv_task_list_head, timeout, OS_FALSE);
            }

            if (OS_SUCCESS == ret)
            {
                msg = (os_mq_msg_t *)_k_task_self()->swap_data;   /* 发送者预留的消息 */
            }
        }
    }

进入本步的条件：msg_queue_head == OS_NULL（队列为空，没有可取的消息）。

• 子分支 A：条件 timeout == OS_NO_WAIT（不等待）→ 直接 ret = OS_EMPTY（返回"队列空"错误码）。
• 子分支 B：条件 timeout != OS_NO_WAIT（等待）：
  • 按 wake_type（OS_MQ_WAKE_TYPE_PRIO → OS_TRUE，否则 → OS_FALSE）把当前任务 k_block_task 挂到接收者等待队列 recv_task_list_head 并让出 CPU，直到有发送者放入消息唤醒它；
  • 被唤醒且 ret == OS_SUCCESS 时，从 _k_task_self()->swap_data 拿到发送者直接转交的消息块（无需入队，见 2.5 的 swap_data 接力机制）。

关键点：msg_queue_head == OS_NULL 是"队列空"的判断——队首指针为 NULL 代表没有已入队的消息。

第三步：队列非空（条件 msg_queue_head != OS_NULL）—— 从队首取出消息

    else
    {
        _k_mq_get_msg_from_queue(mq, msg);
    }

进入本步的条件：第二步的条件不成立，即 msg_queue_head != OS_NULL（队列非空，有消息可取）。

详细解释：_k_mq_get_msg_from_queue(mq, msg) 从队列头部取出一条消息，存到 msg。普通消息按 FIFO（队首是最先入队的）取出；若之前用 send_urgent 插入了队首，则它优先被取出。

第四步：拷贝消息 + 处理缓冲区过小（条件 ret == OS_SUCCESS，且看 buff_size 是否装得下）—— 拷贝到用户缓冲或返回 OS_NOMEM

    if (OS_SUCCESS == ret)
    {
        if (buff_size < msg->msg_len)
        {
            /* When receive buffer size is less than message size, discard this message. */
            OS_KERN_LOG(KERN_ERROR,
                        MQ_TAG,
                        "Recv buff size(%lu) of task(%s) is less than msg size(%lu)",
                        buff_size,
                        _k_task_self()->name,
                        msg->msg_len);
            ret = OS_NOMEM;
        }
        else
        {
            (void)memcpy(buffer, (void *)((uint8_t *)msg + sizeof(os_mq_msg_hdr_t)), msg->msg_len);
            *recv_msg_size = msg->msg_len;
        }
    }

进入本步的条件：ret == OS_SUCCESS（第二步等待成功或第三步成功取出消息，msg 已指向一条有效消息）。

• 子分支 A：条件 buff_size < msg->msg_len（用户缓冲区装不下这条消息）：
  • 打印错误日志（Recv buff size ... is less than msg size ...）；
  • 丢弃该消息，ret = OS_NOMEM（缓冲不足）。
• 子分支 B：条件 buff_size >= msg->msg_len（缓冲区装得下）：
  • memcpy(buffer, (uint8_t*)msg + sizeof(os_mq_msg_hdr_t), msg->msg_len) 把消息体（跳过消息头）拷贝到用户缓冲区；
  • *recv_msg_size = msg->msg_len 记录实际接收的消息大小（写入传出参数）。

关键点：这一步对应前面讲过的"队首接收者不一定要接受这条消息"——若它 buff_size < msg->msg_len，消息被丢弃（OS_NOMEM），不会转给其他接收者。因此调用方必须保证 buffer 容量 ≥ 发送方的 buff_size。

第五步：唤醒等待的发送者 / 释放消息块（条件 send_task_list_head 是否非空）—— 转交消息块给发送者，或释放回空闲池

        if (&mq->send_task_list_head != mq->send_task_list_head.next)
        {
            /* 有发送者阻塞（队列满） */
            block_task = os_list_first_entry(&mq->send_task_list_head, os_task_t, task_node);
            need_sched = k_unblock_task(block_task);
            block_task->swap_data = (os_ubase_t)msg;   /* 把这个消息块交给发送者复用 */
        }
        else
        {
            _k_mq_release_free_msg(mq, msg);   /* 无发送者阻塞，释放消息块回空闲池 */
        }

进入本步的条件：ret == OS_SUCCESS（第四步拷贝成功或丢弃后仍继续走到收尾；注意丢弃时 ret 已是 OS_NOMEM，会跳过本步）。本步内部按"是否有发送者阻塞"分两个子分支。

详细解释：

• 子分支 A：条件 send_task_list_head 非空（有发送者因队列满而阻塞）：取队首发送者，k_unblock_task 唤醒它，并把这个刚取出的空闲消息块地址写入它的 swap_data——发送者被唤醒后直接复用该消息块发送，避免"释放再分配"的往返（与 send 侧对称的 swap_data 接力，见 2.5）。
• 子分支 B：条件 send_task_list_head 为空（无发送者阻塞）：_k_mq_release_free_msg(mq, msg) 把消息块释放回空闲池，供后续发送使用。

关键点 —— 发送与接收的"消息块传递"对称性：send 侧在有接收者等待时把消息块转交给接收者；recv 侧在有发送者等待时把腾出的消息块转交给发送者。两边通过 swap_data 对称复用消息块，避免频繁"释放回池再获取"，是消息队列内存池高效复用的核心设计。若 ret == OS_NOMEM（第四步丢弃消息），本步被跳过，消息块会在丢弃路径后的流程中处理。

关键点 —— 发送与接收的"消息块传递"：os_msgqueue_recv 取出消息后，如果有发送者正因队列满而阻塞，就把这个空闲消息块直接交给发送者（swap_data），让发送者立即用它发送，避免"释放再分配"的往返。这是消息队列内存池复用的高效设计。

使用示例：

char      buf[64];
os_size_t recv_len;
os_err_t  ret;

ret = os_msgqueue_recv(mq_id, buf, sizeof(buf), OS_WAIT_FOREVER, &recv_len);
if (ret == OS_EOK)
{
    os_kprintf("recv %lu bytes: %s\r\n", recv_len, buf);
}

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	获取消息队列名称

os_msgqueue_set_wake_type() —— 设置唤醒策略

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_set_wake_type(os_msgqueue_id msgqueue_id, uint8_t wake_type)
{
    os_err_t       ret;
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_MQ_WAKE_TYPE_PRIO == wake_type) || (OS_MQ_WAKE_TYPE_FIFO == wake_type));

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
    k_kernel_spin_lock();
#else
    OS_KERNEL_ENTER();
#endif

    if (os_list_empty(&mq->recv_task_list_head) && os_list_empty(&mq->send_task_list_head))
    {
        mq->wake_type = wake_type;
        ret           = OS_SUCCESS;
    }
    else
    {
        ret = OS_BUSY;
    }

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    k_kernel_spin_unlock();
    os_spin_unlock_irq(&mq->lock, irq_save);
#else
    OS_KERNEL_EXIT();
#endif

    return ret;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、wake_type 合法）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT((OS_MQ_WAKE_TYPE_PRIO == wake_type) || (OS_MQ_WAKE_TYPE_FIFO == wake_type));

进入本步的条件：调用 os_msgqueue_set_wake_type() 时无条件执行（入口硬性校验）。

• 条件 mq != OS_NULL：队列非空。
• 条件 object_inited == OS_KOBJ_INITED：队列已初始化。
• 条件 wake_type == OS_MQ_WAKE_TYPE_PRIO 或 OS_MQ_WAKE_TYPE_FIFO：唤醒策略参数必须合法。

第二步：修改唤醒策略（条件：两个等待队列都为空）—— 成功修改，否则返回 OS_BUSY

    if (os_list_empty(&mq->recv_task_list_head) && os_list_empty(&mq->send_task_list_head))
    {
        mq->wake_type = wake_type;
        ret           = OS_SUCCESS;
    }
    else
    {
        ret = OS_BUSY;
    }

进入本步的条件：第一步断言通过。

• 子分支 A：条件 recv_task_list_head 和 send_task_list_head 两个等待队列都为空（无任务阻塞）→ mq->wake_type = wake_type 修改成功，返回 OS_SUCCESS。
• 子分支 B：条件两个等待队列任一非空（有任务阻塞）→ 返回 OS_BUSY（队列中有任务已按旧策略排序，不能改）。

关键点：为什么必须在无阻塞任务时才能改？ 因为等待队列的排序方式由 wake_type 决定（PRIO 按优先级、FIFO 按等待先后）。若已有任务按旧策略挂进队列，此时改变策略会导致队列排序混乱。所以只能在"队列完全空闲、无任务等待"时修改。

使用示例：

os_msgqueue_set_wake_type(mq_id, OS_MQ_WAKE_TYPE_FIFO);   /* 在无阻塞任务时设置 */

os_msgqueue_reset() —— 重置消息队列

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_reset(os_msgqueue_id msgqueue_id)
{
    os_mq_msg_t   *msg;
    os_bool_t      need_sched;
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);

    need_sched = OS_FALSE;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    while (1)
    {
        if (OS_NULL == mq->msg_queue_head)
        {
            break;
        }

        _k_mq_get_msg_from_queue(mq, msg);
        _k_mq_release_free_msg(mq, msg);
    }

    if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
    {
        need_sched = OS_TRUE;
    }

    if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
    {
        need_sched = OS_TRUE;
    }

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_unlock_irq(&mq->lock, irq_save);

    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_ENTER();
        OS_KERNEL_EXIT_SCHED();
    }
#else
    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_EXIT_SCHED();
    }
    else
    {
        OS_KERNEL_EXIT();
    }
#endif

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);

进入本步的条件：调用 os_msgqueue_reset() 时无条件执行（入口硬性校验）。

• 条件 mq != OS_NULL：队列非空。
• 条件 object_inited == OS_KOBJ_INITED：队列已初始化。

第二步：清空队列中所有消息（条件：队列非空）—— 循环取出发回空闲池

    while (1)
    {
        if (OS_NULL == mq->msg_queue_head)
        {
            break;
        }
        _k_mq_get_msg_from_queue(mq, msg);
        _k_mq_release_free_msg(mq, msg);
    }

进入本步的条件：第一步断言通过。本步是 while(1) 循环。

• 子分支 A：条件 msg_queue_head == OS_NULL（队列已空）→ break 跳出循环。
• 子分支 B：条件 msg_queue_head != OS_NULL（队列还有消息）→ _k_mq_get_msg_from_queue 取出一条，_k_mq_release_free_msg 释放回空闲池，循环继续，直到队列清空。

关键点：这一步把所有已入队消息逐个取出发回空闲池，使队列回到"空且全部消息块可用"的初始状态。

第三步：唤醒所有阻塞的收发任务（条件：等待队列非空）—— 触发调度

    if (OS_TRUE == k_cancel_all_blocked_task(&mq->send_task_list_head))
    {
        need_sched = OS_TRUE;
    }
    if (OS_TRUE == k_cancel_all_blocked_task(&mq->recv_task_list_head))
    {
        need_sched = OS_TRUE;
    }
    if (OS_TRUE == need_sched)
    {
        OS_KERNEL_EXIT_SCHED();
    }
    else
    {
        OS_KERNEL_EXIT();
    }
    return OS_SUCCESS;

进入本步的条件：第二步已清空队列。

• 子分支 A：条件 send_task_list_head 非空 → 唤醒所有阻塞的发送者，返回 OS_TRUE 则 need_sched = OS_TRUE。
• 子分支 B：条件 recv_task_list_head 非空 → 唤醒所有阻塞的接收者，返回 OS_TRUE 则 need_sched = OS_TRUE。
• 子条件 need_sched == OS_TRUE（有唤醒）→ OS_KERNEL_EXIT_SCHED() 触发调度；否则仅退出临界区。

关键点：重置会清空队列并唤醒所有阻塞的收发任务，让它们从等待中恢复（收到 OS_SUCCESS 但队列已空，实际取不到消息）。与 destroy 类似，都是先清等待队列再处理。

使用示例：

os_err_t ret = os_msgqueue_reset(mq_id);   /* 清空所有消息并唤醒阻塞任务 */
if (ret != OS_EOK)
{
    os_kprintf("reset failed\r\n");
}

os_msgqueue_is_empty() —— 检查队列是否为空

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_is_empty(os_msgqueue_id msgqueue_id, os_bool_t *empty_flag)
{
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != empty_flag);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    if (OS_NULL == mq->msg_queue_head)
    {
        *empty_flag = OS_TRUE;
    }
    else
    {
        *empty_flag = OS_FALSE;
    }

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_unlock_irq(&mq->lock, irq_save);
#else
    OS_KERNEL_EXIT();
#endif

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、empty_flag 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != empty_flag);

进入本步的条件：调用 os_msgqueue_is_empty() 时无条件执行（入口硬性校验）。

• mq != OS_NULL：队列非空。
• object_inited == OS_KOBJ_INITED：队列已初始化。
• empty_flag != OS_NULL：传出参数必须非空（函数要写它）。

第二步：判断队首是否为空（条件：msg_queue_head 是否非空）—— 写入结果

    if (OS_NULL == mq->msg_queue_head)
    {
        *empty_flag = OS_TRUE;
    }
    else
    {
        *empty_flag = OS_FALSE;
    }
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。

• 子分支 A：条件 msg_queue_head == OS_NULL（队首为空 → 队列空）→ *empty_flag = OS_TRUE。
• 子分支 B：条件 msg_queue_head != OS_NULL（队首非空 → 队列非空）→ *empty_flag = OS_FALSE。

关键点：判断"是否为空"的依据是队首指针 msg_queue_head 是否为 NULL（不是 used_msgs）。返回恒为 OS_SUCCESS，结果通过传出参数 empty_flag 告知调用方。

os_msgqueue_is_full() —— 检查队列是否已满

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_is_full(os_msgqueue_id msgqueue_id, os_bool_t *full_flag)
{
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != full_flag);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    if (OS_NULL == mq->msg_queue_free)
    {
        *full_flag = OS_TRUE;
    }
    else
    {
        *full_flag = OS_FALSE;
    }

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_unlock_irq(&mq->lock, irq_save);
#else
    OS_KERNEL_EXIT();
#endif

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、full_flag 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != full_flag);

进入本步的条件：调用 os_msgqueue_is_full() 时无条件执行（入口硬性校验）。

• mq != OS_NULL：队列非空。
• object_inited == OS_KOBJ_INITED：队列已初始化。
• full_flag != OS_NULL：传出参数必须非空。

第二步：判断空闲槽位是否为 NULL（条件：msg_queue_free 是否非空）—— 写入结果

    if (OS_NULL == mq->msg_queue_free)
    {
        *full_flag = OS_TRUE;
    }
    else
    {
        *full_flag = OS_FALSE;
    }
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。

• 子分支 A：条件 msg_queue_free == OS_NULL（没有空闲消息槽位 → 队列满）→ *full_flag = OS_TRUE。
• 子分支 B：条件 msg_queue_free != OS_NULL（还有空闲槽位 → 队列未满）→ *full_flag = OS_FALSE。

关键点：判断"是否已满"的依据是空闲消息指针 msg_queue_free 是否为 NULL（与 send 里 msg_queue_free == OS_NULL 判断队列满一致）。返回恒为 OS_SUCCESS。

os_msgqueue_get_queue_depth() —— 获取队列容量

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_get_queue_depth(os_msgqueue_id msgqueue_id, uint16_t *msgqueue_depth)
{
    os_msgqueue_t *mq;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != msgqueue_depth);

    *msgqueue_depth = mq->queue_depth;

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、msgqueue_depth 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != msgqueue_depth);

进入本步的条件：调用 os_msgqueue_get_queue_depth() 时无条件执行（入口硬性校验）。msgqueue_depth 是传出参数，必须非空。

第二步：写入队列容量（无条件执行）—— 返回 queue_depth

    *msgqueue_depth = mq->queue_depth;
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。把 mq->queue_depth（队列总容量，即最多能容纳的消息数）写入传出参数，返回 OS_SUCCESS。

os_msgqueue_get_max_msg_size() —— 获取单条消息最大大小

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_get_max_msg_size(os_msgqueue_id msgqueue_id, os_size_t *max_msg_size)
{
    os_msgqueue_t *mq;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT((OS_NULL != mq));
    OS_ASSERT((OS_KOBJ_INITED == mq->object_inited));
    OS_ASSERT((OS_NULL != max_msg_size));

    *max_msg_size = mq->max_msg_size;

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、max_msg_size 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT((OS_NULL != mq));
    OS_ASSERT((OS_KOBJ_INITED == mq->object_inited));
    OS_ASSERT((OS_NULL != max_msg_size));

进入本步的条件：调用 os_msgqueue_get_max_msg_size() 时无条件执行（入口硬性校验）。max_msg_size 是传出参数，必须非空。

第二步：写入单条消息最大大小（无条件执行）—— 返回 max_msg_size

    *max_msg_size = mq->max_msg_size;
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。把 mq->max_msg_size（每条消息的最大字节数）写入传出参数，返回 OS_SUCCESS。

os_msgqueue_get_used_msgs() —— 获取已用消息数

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_get_used_msgs(os_msgqueue_id msgqueue_id, uint16_t *used_msgs)
{
    os_msgqueue_t *mq;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != used_msgs);

    *used_msgs = mq->used_msgs;

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、used_msgs 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != used_msgs);

进入本步的条件：调用 os_msgqueue_get_used_msgs() 时无条件执行（入口硬性校验）。used_msgs 是传出参数，必须非空。

第二步：写入已用消息数（无条件执行）—— 返回 used_msgs

    *used_msgs = mq->used_msgs;
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。把 mq->used_msgs（当前已存入队列的消息数）写入传出参数，返回 OS_SUCCESS。

os_msgqueue_get_unused_msgs() —— 获取剩余可用消息数

完整源码（kernel/source/os_mq.c，与项目一致）：

os_err_t os_msgqueue_get_unused_msgs(os_msgqueue_id msgqueue_id, uint16_t *unused_msgs)
{
    os_msgqueue_t *mq;

    OS_KERNEL_INIT();

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != unused_msgs);

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_lock_irq(&mq->lock, &irq_save);
#else
    OS_KERNEL_ENTER();
#endif

    *unused_msgs = mq->queue_depth - mq->used_msgs;

#if defined(OS_USING_SMP) && !defined(OS_IPC_MONOLOCK_MODE)
    os_spin_unlock_irq(&mq->lock, irq_save);
#else
    OS_KERNEL_EXIT();
#endif

    return OS_SUCCESS;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化、unused_msgs 传出参数非空）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);
    OS_ASSERT(OS_NULL != unused_msgs);

进入本步的条件：调用 os_msgqueue_get_unused_msgs() 时无条件执行（入口硬性校验）。unused_msgs 是传出参数，必须非空。

第二步：写入剩余可用消息数（无条件执行）—— 返回 queue_depth - used_msgs

    *unused_msgs = mq->queue_depth - mq->used_msgs;
    return OS_SUCCESS;

进入本步的条件：第一步断言通过。计算 容量 - 已用 = 剩余可用 写入传出参数，返回 OS_SUCCESS。

关键点：与 get_used_msgs 是互补关系——used + unused = queue_depth（总容量）。

os_msgqueue_get_name() —— 获取消息队列名称

完整源码（kernel/source/os_mq.c，与项目一致）：

const char *os_msgqueue_get_name(os_msgqueue_id msgqueue_id)
{
    os_msgqueue_t *mq;

    mq = OS_TYPE_CONVERT(os_msgqueue_t *, msgqueue_id);

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);

    return mq->name;
}

分步详解：

第一步：准备与参数校验（条件：队列非空、已初始化）—— 不满足则断言崩溃

    OS_ASSERT(OS_NULL != mq);
    OS_ASSERT(OS_KOBJ_INITED == mq->object_inited);

进入本步的条件：调用 os_msgqueue_get_name() 时无条件执行（入口硬性校验）。mq 非空、队列已初始化。

第二步：返回名称字符串指针（无条件执行）

    return mq->name;

进入本步的条件：第一步断言通过。直接返回 mq->name（消息队列名称字符串指针）。注意：这是 2.8 中唯一返回 const char *（非 OS_SUCCESS）的函数，调用方拿到的是指向队列内部 name 数组的指针，只读使用。

📝 本节小结

核心要点回顾

本节围绕消息队列机制展开，核心要点如下：

1. 消息队列的本质

• 一种变长数据传递机制，以数据拷贝方式通信（发送方和接收方各自持有独立副本）。
• 内部维护内存池（msg_pool）存放消息，通过队首/队尾/空闲指针管理消息块。

2. 两种创建方式

方式	函数	特点
动态创建	os_msgqueue_create_dynamic	灵活，从堆分配控制块和内存池（需 OS_USING_HEAP）
静态创建	os_msgqueue_create_static	确定性，用户提供控制块和内存池（无碎片）

3. 发送 / 接收核心行为

操作	队列有空间/消息	队列满/空
send	拷贝数据，放入队列或转交给等待的接收者	等待空闲槽位 或 返回 OS_FULL
recv	从队首取消息，拷贝到用户缓冲区	等待消息 或 返回 OS_EMPTY

4. 紧急发送 send_urgent

• 插入队首，优先被接收者取出（普通消息插队尾）。
• 其余逻辑与普通 send 完全相同。

5. 高效的内存池复用

• 发送时若有接收者阻塞，消息直接转交给接收者（swap_data），不入队。
• 接收后若有发送者阻塞，把空闲消息块直接交给发送者复用，避免"释放再分配"。

6. 中断限制

• 消息队列不能在中断上下文中使用（发送/接收都可能阻塞，与信号量的 post 不同）。

关键易错点速查

易错点	正确做法
在中断中使用消息队列	禁止，发送/接收都可能阻塞；中断中应用信号量 post 等不阻塞接口
发送的消息大小超过 max_msg_size	断言失败，发送前需检查 buff_size <= mq->max_msg_size
接收缓冲区太小	返回 OS_NOMEM 且丢弃该消息，应保证缓冲区 ≥ 消息大小
忽略返回值	OS_FULL/OS_EMPTY/OS_TIMEOUT 需处理，避免误以为成功
忘记检查 recv_msg_size 传出参数	应为非空指针，否则断言失败
用消息队列传大块数据	每次 send 都要拷贝，大块数据可用邮箱传指针更高效

下一步

消息队列用于传递变长数据。接下来请学习下一小节：

• 4.3.5 邮箱：了解如何用邮箱传递指针值，实现更高效的数据通信