RT-Thread之消息队列使用示例

消息队列（Message Queue）是一种异步通信机制，其核心功能是：

• 解耦生产者和消费者：发送方和接收方无需同时在线
• 缓冲与流量整形：应对瞬时流量过载
• 优先级处理：支持紧急消息插队

典型案例：
在RT-Thread中，中断服务程序（ISR）通过rt_mq_send非阻塞发送传感器数据，消费者线程按需从队列读取数据，避免ISR因等待导致实时性下降。

一、消息队列API函数

1、消息队列的创建

• 动态创建消息队列

rt_mq_t rt_mq_create(const char *name,rt_size_t   msg_size,rt_size_t   max_msgs,rt_uint8_t  flag)

• 静态创建消息队列

rt_err_t rt_mq_init(rt_mq_t     mq,const char *name,void       *msgpool,rt_size_t   msg_size,rt_size_t   pool_size,rt_uint8_t  flag)

2、消息的发送

发送消息时，从空闲消息链表取一个空闲消息块，把消息复制到该消息块，然后把消息块挂到消息队列尾部。消息的发送有以下三种：

• 直接发送消息

队列中有空闲消息块时，才能成功发送消息，否则返回错误。

rt_err_t rt_mq_send(rt_mq_t mq, const void *buffer, rt_size_t size)

• 等待发送消息

如果队列中没有可用的空闲消息块，会根据timeout参数等待，超时后才返回错误。

rt_err_t rt_mq_send_wait(rt_mq_t     mq,const void *buffer,rt_size_t   size,rt_int32_t  timeout)

• 紧急发送消息

它会把消息块放在消息队列的头部，以便这个消息能被第1时间读取。

rt_err_t rt_mq_urgent(rt_mq_t mq, const void *buffer, rt_size_t size)

3、消息的接收

当队列有消息时，使用收消息函数，可以从队列接收消息；如果没有消息，根据指定的 timeout 参数等待，直到超时结束。

rt_err_t rt_mq_recv(rt_mq_t    mq,void      *buffer,rt_size_t  size,rt_int32_t timeout)

4、消息队列的脱离/删除

• 删除使用 rt_mq_create()创建的队列

rt_err_t rt_mq_delete(rt_mq_t mq)

• 脱离使用 rt_mq_init()初始化的队列

rt_err_t rt_mq_detach(rt_mq_t mq)

二、创建消息队列示例

1、创建消息队列

/* 使用静态创建（确定内存占用）	*/
struct rt_messagequeue mq_static;
uint8_t msgpool[96];  /* 消息数量计算：96/(RT_ALIGN(2,4)+4)=12条消息 */
rt_mq_init(&mq_static,         /* 消息队列对象的句柄 */"smq",              /* 消息队列的名字 */msgpool,            /* 内存池指向的地址 */2,                  /* 每一个消息占2字节 */96,                 /* 内存池的大小 */RT_IPC_FLAG_FIFO);  /* 选择线程唤醒的顺序方式 *//* 使用动态创建（灵活调整） */
rt_mq_t mq_dynamic = rt_mq_create("dmq",               /* 消息队列名称 */2,                  /* 每一个消息大小 */12,                 /* 消息数量 */RT_IPC_FLAG_FIFO);  /* 选择线程唤醒的顺序方式 */

2、完整示例

#include "thread_task.h"
#include "main.h"
#include <stdio.h>      
#include "rtthread.h"
#include <rthw.h>/******************************************** 线程 1 ******************************************************/
#define THREAD_1_PRIORITY  		4           /* 进程优先级 */
#define THREAD_1_STACK_SIZE		512         /* 进程栈空间大小 */
#define THREAD_1_TIMESLICE		10           /* 进程执行时间片个数 */
static struct rt_thread *thread_1_handle;        /* 进程句柄 *//******************************************** 线程 2 ******************************************************/
#define THREAD_2_PRIORITY  		5           /* 进程优先级 */
#define THREAD_2_STACK_SIZE		512         /* 进程栈空间大小 */
#define THREAD_2_TIMESLICE		10           /* 进程执行时间片个数 */
static struct rt_thread *thread_2_handle;    /* 进程句柄 *//* 消息队列 */
rt_mq_t mq_test;/*** @brief   线程1入口函数* @param   无* @retval  无*/
void thread_1_entry(void* param)
{uint8_t buffer[2];while(1){rt_mq_recv(mq_test, buffer, 2, 0xffffffff);HAL_GPIO_TogglePin(GPIOC, LED1_Pin);rt_thread_delay(4000);     /* 精准延时1000时间片 */}
}/*** @brief   线程2入口函数* @param   无* @retval  无*/
void thread_2_entry(void* param)
{uint8_t text[2] = {0,1};while(1){text[0] ++;rt_mq_send_wait(mq_test, text, 2, 0xffff);HAL_GPIO_TogglePin(GPIOC, LED2_Pin);rt_thread_delay(200);    }
}/*** @brief   动态创建线程任务并启动* @param   无* @retval  无*/
void ThreadStart(void)
{rt_base_t level = rt_hw_interrupt_disable();/* 动态创建线程 */thread_1_handle = rt_thread_create("thread_1",				/* 线程句柄名称*/thread_1_entry,			/* 函数入口 */RT_NULL,				/* 入口函数参数 */THREAD_1_STACK_SIZE,	/* 线程栈大小 */THREAD_1_PRIORITY,		/* 线程优先级 */THREAD_1_TIMESLICE  	/* 线程时间片大小 */);rt_thread_startup(thread_1_handle); /* 启动线程 */thread_2_handle = rt_thread_create("thread_2",				/* 线程句柄名称*/thread_2_entry,			/* 函数入口 */RT_NULL,				/* 入口函数参数 */THREAD_2_STACK_SIZE,	/* 线程栈大小 */THREAD_2_PRIORITY,		/* 线程优先级 */THREAD_2_TIMESLICE  	/* 线程时间片大小 */);rt_thread_startup(thread_2_handle); /* 启动线程 *//* 动态创建一个消息队列 */mq_test = rt_mq_create("mq_test",              /* 消息队列名称 */2,                      /* 消息大小 */12,                     /* 消息数量 */RT_IPC_FLAG_FIFO        /* 选择线程唤醒的顺序方式 */);rt_hw_interrupt_enable(level);
}

执行流程如下:

1. 初始阶段

• 线程1立即阻塞等待消息（队列空）
• 线程2开始高频发送（每200 ticks发送1次）

2. 队列填满阶段
   • 线程2发送12条消息后队列满 → rt_mq_send_wait()开始阻塞
   • 系统进入双阻塞状态（线程1和2均挂起）
3. 恢复同步阶段
   • 线程1的rt_thread_delay(4000)到期后：
     • 消费1条消息 → 释放1个队列空间
     • 唤醒线程2继续发送
   • 稳定状态：每4000 ticks消费1条，线程2每200 ticks补充1条

---