线程间通信_消息队列

Tutorial: 卓越工程师 Category: 嵌入式 Published: 2026-10-23 11:05:29 Views: 1 Likes: 0 Comments: 0

260.2.11线程间通信_消息队列

#include <rtthread.h> // 引入 RT-Thread 的线程、消息队列、延时和内核打印接口
#define DBG_TAG "main" // 设置 rtdbg 日志标签为 main
#define DBG_LVL DBG_LOG // 设置 rtdbg 日志级别
#include <rtdbg.h> // 引入 LOG_E 等 RT-Thread 调试日志宏
rt_thread_t th1_ptr,th2_ptr; // 定义两个动态线程句柄：th1_ptr 用于接收，th2_ptr 用于发送
rt_mq_t mq_ptr; // 定义动态消息队列句柄，后续保存 rt_mq_create() 的返回值
struct msg // 定义本实验每条消息采用的业务结构体类型
{ // struct msg 结构体定义开始
    char str[100]; // 分配 100 字节字符数组，用于保存以 '\0' 结尾的字符串
    uint8_t len; // 用一个 8 位无符号整数记录字符串中可见字符的数量
}; // struct msg 结构体定义结束
char str1[]="12345678"; // 定义具有全局生命周期的第一个待发送字符串，长度为 8
char str2[]="67890"; // 定义具有全局生命周期的第二个待发送字符串，长度为 5
void th1_recv(void *parameter) // 定义线程 th1 的入口函数，负责从消息队列接收结构体副本
{ // th1_recv 函数体开始
    rt_err_t ret; // 保存课程源码中 rt_mq_recv() 返回的接收长度或负错误码
    struct msg mq_msg; // 在接收线程栈中创建一个结构体，作为消息接收缓冲区
    while(1) // 接收线程永久循环，持续等待并处理新消息
    { // 接收循环体开始
        ret = rt_mq_recv(mq_ptr, &mq_msg,sizeof(struct msg), RT_WAITING_FOREVER); // 永久等待一条消息，并把数据复制到 mq_msg
        if(ret > 0) // 返回正数表示成功收到相应字节数的数据
        { // 接收成功分支开始
            rt_kprintf("rt_mq_recv is %s the len is %d!\r\n" ,mq_msg.str,mq_msg.len); // 打印结构体中的字符串及其字符长度
        }else{ // 接收失败分支开始
            LOG_E("mq rt_mq_recv failed!\r\n"); // 记录消息队列接收失败信息
        } // 接收结果判断结束
    } // 接收循环体结束
} // th1_recv 函数体结束
void th2_send(void *parameter) // 定义线程 th2 的入口函数，负责组装并发送结构体消息
{ // th2_send 函数体开始
    rt_err_t ret; // 保存消息队列发送接口返回的 RT-Thread 错误码
    struct msg mq_msg; // 在发送线程栈中创建待发送结构体
    while(1) // 发送线程永久循环，交替发送两组内容
    { // 发送循环体开始
        strcpy(mq_msg.str,str1); // 把字符串 str1 连同结尾 '\0' 复制到结构体字符数组
        mq_msg.len=strlen(mq_msg.str); // 计算 str1 的可见字符数并保存到 len，本次为 8
        ret = rt_mq_send(mq_ptr, &mq_msg, sizeof(struct msg)); // 把整个 mq_msg 的字节副本发送到消息队列
        if(ret != RT_EOK) // 判断第一条结构体消息是否发送失败
        { // 第一次发送失败分支开始
            LOG_E("mb rt_mb_send failed!\r\n"); // 保留课程源码的日志文字；这里实际调用的是消息队列发送接口
        } // 第一次发送结果判断结束
        rt_thread_mdelay(1000); // 让发送线程延时约 1000 ms
        strcpy(mq_msg.str,str2); // 把字符串 str2 连同结尾 '\0' 复制到同一个结构体字符数组
        mq_msg.len=strlen(mq_msg.str); // 计算 str2 的可见字符数并保存到 len，本次为 5
        ret = rt_mq_send(mq_ptr, &mq_msg, sizeof(struct msg)); // 再把整个 mq_msg 的当前字节内容复制进队列
        if(ret != RT_EOK) // 判断第二条结构体消息是否发送失败
        { // 第二次发送失败分支开始
            LOG_E("mb rt_mb_send failed!\r\n"); // 保留课程源码由邮箱例程复制而来的错误日志文字
        } // 第二次发送结果判断结束
        rt_thread_mdelay(1000); // 再延时约 1000 ms，然后进入下一轮发送
    } // 发送循环体结束
} // th2_send 函数体结束
int main(void) // RT-Thread 用户主线程入口函数
{ // main 函数体开始
    int count = 1; // 初始化课程用于维持主线程循环的计数变量
    mq_ptr = rt_mq_create("mq", sizeof(struct msg),4, RT_IPC_FLAG_FIFO); // 创建单条上限为一个 msg、最多缓存 4 条消息的动态队列
    if(mq_ptr == RT_NULL) // 判断消息队列控制块或消息池是否分配失败
    { // 消息队列创建失败分支开始
        LOG_E("mq_ptr rt_mq_create failed!\r\n"); // 记录动态消息队列创建失败信息
        return -RT_ENOMEM; // 返回内存不足错误，停止后续线程创建
    } // 消息队列创建结果判断结束
    th1_ptr = rt_thread_create("th1", th1_recv, NULL, 1024, 7, 10); // 创建栈为 1024 字节、优先级为 7、时间片为 10 tick 的接收线程
    if(th1_ptr == RT_NULL) // 判断接收线程是否创建失败
    { // 接收线程创建失败分支开始
        LOG_E("th1_recv rt_thread_create failed!\r\n"); // 记录接收线程创建失败信息
        return -RT_ENOMEM; // 返回内存不足错误，停止后续初始化
    } // 接收线程创建结果判断结束
    rt_thread_startup(th1_ptr); // 启动接收线程；队列为空时它会阻塞等待
    th2_ptr = rt_thread_create("th2", th2_send, NULL, 1024, 8, 10); // 创建栈为 1024 字节、优先级为 8、时间片为 10 tick 的发送线程
    if(th2_ptr == RT_NULL) // 判断发送线程是否创建失败
    { // 发送线程创建失败分支开始
        LOG_E("th2_send rt_thread_create failed!\r\n"); // 记录发送线程创建失败信息
        return -RT_ENOMEM; // 返回内存不足错误，停止后续初始化
    } // 发送线程创建结果判断结束
    rt_thread_startup(th2_ptr); // 启动发送线程，使其开始交替投递两组结构体数据
    while (count++) // 课程用自增变量维持主线程循环，通常不会主动退出
    { // 主线程循环体开始
        //LOG_D("Hello RT-Thread!"); // 课程把周期日志注释掉，避免干扰消息队列实验输出
        rt_thread_mdelay(1000); // 主线程每次延时约 1000 ms，让其他线程和 shell 正常运行
    } // 主线程循环体结束
    return RT_EOK; // 理论上的正常返回值；本例通常不会执行到这里
} // main 函数体结束

源码说明： 上方是课程配套 10MQ 工程最终的 applications/main.c 业务代码。除去文件头版权说明和原有说明性注释后，结构体、变量、字符串、函数、参数、调用顺序、错误日志以及被注释掉的 LOG_D 均按课程源码保留，只增加逐行初学者注释；没有补写 RT-Thread 消息队列的库函数实现。两处发送失败日志仍写着 mb rt_mb_send failed，这是由上一节邮箱工程复制代码后遗留的提示文字，实际调用不受影响，因此按“源码原样保留”要求没有改写。

一、线程间通信

线程间通信用于让不同线程传递数据或通知。本节在上一课邮箱的基础上学习消息队列：发送线程把一段数据复制进内核队列，接收线程再从队列中取走并复制到自己的缓冲区。发送和接收不必在同一时刻发生，未被接收的数据可以暂存在有限数量的消息槽中。

1.1 消息队列

消息队列可以理解为邮箱的扩展。邮箱每封邮件只能保存一个机器字，而消息队列的一条消息可以包含多个字节，因此能直接传递数组、结构体或协议数据帧。

“可以传递不定长数据”不能理解为每条消息没有长度上限。准确含义是：

  • 创建消息队列时，用 msg_size 固定该队列单条消息的最大长度。
  • 每次发送时再用 size 指定本条消息的实际长度，只要满足 0 < size <= msg_size 即可。
  • 不同次发送可以使用不同的实际长度，但都不能超过创建时的上限。
  • 队列会保存每条消息的实际长度，RT-Thread 5.0.2 的接收接口会返回实际复制的字节数。

本例没有发送指针，而是直接复制 struct msg：

th2_send 的局部 mq_msg          内核消息队列 mq（最多 4 条）          th1_recv 的局部 mq_msg
[str + len] ──rt_mq_send──> [副本1 | 副本2 | 空 | 空] ──rt_mq_recv──> [str + len]
              复制结构体字节                                  再复制到接收缓冲区

因此，th2_send 下一轮修改自己的局部变量不会改变已经入队的消息；这与上一课邮箱只保存指针地址有本质区别。

本节开头还补充了上一课未实验的 rt_mb_urgent()：它把紧急邮件放到邮箱队头，使接收者优先取得。消息队列也有对应的 rt_mq_urgent()，原理相同；紧急发送只是改变队列位置，并不会突破容量限制。课程本节只口头介绍，没有把紧急发送写进最终业务代码，因此文首源码不作补充。

1.1.1 rt_mq_create函数

动态创建消息队列的接口为：

rt_mq_t rt_mq_create(const char *name,
                     rt_size_t msg_size,
                     rt_size_t max_msgs,
                     rt_uint8_t flag);
参数本例准确含义
name"mq"内核对象名称，用于对象管理和调试
msg_sizesizeof(struct msg)每条消息允许占用的最大字节数
max_msgs4队列最多同时缓存 4 条尚未取走的消息
flagRT_IPC_FLAG_FIFO多个线程阻塞在该 IPC 对象上时，按等待先后排列

这里再次要区分两种“FIFO”：

  • 普通消息本身由队尾进入、从队头取出，所以按先发送先接收排列；rt_mq_urgent() 可以例外地插到队头。
  • RT_IPC_FLAG_FIFO 控制的是多个阻塞等待线程的排列和唤醒顺序，不是用来开启消息的先进先出属性。若改为 RT_IPC_FLAG_PRIO，等待线程按优先级排列。

struct msg 包含两个成员：

struct msg
{
    char str[100];
    uint8_t len;
};
  • str[100] 最多安全容纳 99 个普通字符和末尾的 \0。
  • len 保存 strlen() 得到的可见字符数，不包含结尾 \0。
  • sizeof(struct msg) 是整个结构体占用的字节数，不等于当前字符串长度；队列据此为每个消息槽预留空间。
  • 内核还会按 RT_ALIGN_SIZE 对消息数据区进行对齐，并为每个槽附加内部消息头，所以实际分配的消息池会大于简单的 sizeof(struct msg) × 4。

本例使用动态对象，rt_mq_create() 会从内核堆分配消息队列控制块和消息池，任一分配失败都返回 RT_NULL。消息队列也有静态与动态两套配对接口：

对象形式建立释放内存来源
动态消息队列rt_mq_create()rt_mq_delete()内核动态堆
静态消息队列rt_mq_init()rt_mq_detach()调用者提供控制块和消息池

课程提到消息队列具有“创建、发送、接收、删除”四类操作，但本例对象需要在系统运行期间一直存在，最终代码没有调用 rt_mq_delete()，这里也不能擅自补入。真正删除前，必须确保其他执行路径不会继续访问已经失效的句柄。

配套 10MQ 工程使用 RT-Thread 5.0.2，系统 tick 频率为 1000 Hz，启用了消息队列、动态堆和 UART1 控制台。本实验没有新增外部硬件或电路；串口只负责输出观察结果。

1.1.2 rt_mq_send函数

普通消息发送接口为：

rt_err_t rt_mq_send(rt_mq_t mq, const void *buffer, rt_size_t size);
参数本例含义
mqmq_ptr目标消息队列句柄
buffer&mq_msg待复制数据的起始地址，不是结构体对象本身的值
sizesizeof(struct msg)本次需要复制进队列的字节数

发送时，内核从空闲消息槽链表中取出一个槽，把 buffer 指向的前 size 个字节复制进去，然后将该槽接到消息队列尾部。因为执行的是内容复制，所以本例可以安全地发送线程局部变量 mq_msg；rt_mq_send() 返回后，队列中的副本不再依赖这个局部变量的地址。

返回值需要区分：

  • RT_EOK：发送成功。
  • -RT_EFULL：队列的 4 个槽都已占用；普通 rt_mq_send() 不等待空位，立即失败。
  • -RT_ERROR：本次 size 超过创建时的 msg_size 等发送错误。

若希望队列满时让线程等待空位，可以使用带超时的 rt_mq_send_wait()。非阻塞发送可在线程或中断上下文使用，但接收等待不能放在中断服务程序中。

发送线程组装第一条消息的过程是：

strcpy(mq_msg.str, str1);
mq_msg.len = strlen(mq_msg.str);
rt_mq_send(mq_ptr, &mq_msg, sizeof(struct msg));

C 数组不能在定义完成后用 = 整体赋予另一个字符串，因此课程使用 strcpy() 把字符逐个复制，并包含末尾 \0。strlen() 只计算 \0 之前的字符数，所以两组数据的 len 分别为 8 和 5。

要注意以下边界：

  • strcpy() 不知道目标数组容量。如果源字符串超过 99 个字符，就会越界写坏栈；本例两个固定字符串长度足够小，所以演示可正常运行。
  • strlen() 返回 size_t，课程把结果保存到 uint8_t。本例最大有效长度只有 99，不会截断；若换成更大缓冲区，应重新考虑字段类型。
  • 本例每次发送 sizeof(struct msg)，而 strcpy() 只写到 \0 为止，str 后方未使用区域可能保留旧的栈字节。接收端不会打印这些尾部字节，但整结构体发送会复制它们并消耗固定带宽。产品代码可通过初始化结构体或设计更紧凑的消息格式使内容确定；文首仍保留课程原样。
  • 两处失败日志中的 mb、rt_mb_send 只是显示文本写错，实际执行的仍是 rt_mq_send()。

发送线程先发送 "12345678"，延时约 1 秒，再发送 "67890"，再延时约 1 秒并循环。

1.1.3 rt_mq_recv函数

RT-Thread 5.0.2 中，接收接口的实际声明是：

rt_ssize_t rt_mq_recv(rt_mq_t mq,
                      void *buffer,
                      rt_size_t size,
                      rt_int32_t timeout);
参数本例含义
mqmq_ptr从哪个消息队列取数据
buffer&mq_msg接收数据要复制到的缓冲区地址
sizesizeof(struct msg)接收缓冲区允许写入的最大字节数
timeoutRT_WAITING_FOREVER队列为空时永久等待

收到消息后，内核取出队头消息，将“该消息实际长度”和“接收缓冲区容量”中的较小值复制到 buffer，并返回实际复制的字节数。队列为空时，本例会挂起 th1，有新消息到达才唤醒，因此接收循环不需要额外延时，也不会空转占用 CPU。

在这个版本中应按下面的规则判断返回值：

  • 返回正数：成功，数值是实际复制的字节数。
  • 返回负数：失败，例如非阻塞接收时队列为空或等待超时会返回 -RT_ETIMEOUT。
  • 本接口要求请求大小非零，所以正常成功不会返回 0。

因此课程把邮箱例程中的 ret == RT_EOK 改为 ret > 0 是正确的。源码仍把 ret 声明成 rt_err_t，在本工程平台上能保存这些返回值；从接口语义和声明一致性看，正式代码使用 rt_ssize_t 更清楚，但不能据此擅自改写课程源码。

接收成功后，%s 打印 mq_msg.str，%d 打印经过整数提升的 mq_msg.len。正常输出交替为：

rt_mq_recv is 12345678 the len is 8!
rt_mq_recv is 67890 the len is 5!
rt_mq_recv is 12345678 the len is 8!
rt_mq_recv is 67890 the len is 5!
1.1.4 本节代码的调度过程
  1. main() 创建单条上限为 sizeof(struct msg)、深度为 4 的动态消息队列。
  2. main() 创建并启动优先级为 7 的接收线程 th1。
  3. th1 发现队列为空，在 rt_mq_recv() 处永久阻塞。
  4. main() 恢复运行，创建并启动优先级为 8 的发送线程 th2。
  5. th2 把第一份结构体内容复制进队列；等待中的 th1 被唤醒。
  6. th1 把队头数据复制到自己的局部结构体并打印，随后再次阻塞等待。
  7. th2 每隔约 1 秒交替发送两份内容，th1 随到随取。

RT-Thread 中优先级数值越小，线程优先级越高，因此 th1 的优先级 7 高于 th2 的优先级 8。只要 th1 被新消息唤醒，它就可以抢占发送线程完成接收和打印。

1.1.5 消息队列与邮箱的选择
对比项邮箱消息队列
单条容量一个 rt_ubase_t 机器字；本课程平台为 4 字节创建时指定最大字节数
入队内容直接保存一个整数或地址值把发送缓冲区中的字节复制到内部消息槽
大数据常见做法发送指针，应用负责目标对象生命周期直接发送结构体或数据块副本
时间与空间开销小，固定复制一个机器字随实际消息长度增加，消息池也更大
适用场景状态码、句柄、指针数据帧、结构体、需要保存快照的数据

消息队列避免了“发送局部结构体地址后函数返回导致悬空指针”的问题，但结构体内部若仍包含指针，队列复制的仍只是指针值，不会递归复制它指向的数据。此时那块外部数据的生命周期仍需单独管理。

1.1.6 教学代码中的简化点
  • 两个线程入口的 parameter 均未使用，这是 RT-Thread 线程入口统一函数签名带来的正常现象。
  • 源码没有检查 rt_thread_startup() 的返回值，也没有为中途创建失败设计逆序资源清理，这是突出消息队列流程的教学简化。
  • while (count++) 的有符号 int 长期自增最终会发生溢出，不宜作为产品代码的永久循环条件；文首仍按课程源码保留。
  • LOG_D("Hello RT-Thread!") 被注释掉，是为了避免主线程每秒日志干扰实验输出。
  • 多个发送线程和接收线程可安全共用消息队列内部机制，但一条消息只会由一个接收线程取走，并非广播给所有线程。

核对依据： 本节代码、版本、配置和运行结果取自课程视频、转录文本及配套 10MQ 工程；消息块复制、长度限制、紧急消息、等待线程排序、错误码和 RT-Thread 5.0.2 的接收返回值已与该版本的 ipc.c、rtthread.h、rtdef.h，以及官方线程间通信文档交叉核对。官方网页中的部分旧版函数原型仍把 rt_mq_recv() 写成 rt_err_t，本笔记涉及本课程工程时以配套 5.0.2 头文件和内核源码中的 rt_ssize_t 声明为准。

Prev: 线程间通信_邮箱 Next: 线程间通信_信号