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

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

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

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

1、消息队列的创建

• 动态创建消息队列
  

1. rt_mq_t rt_mq_create(const char *name,
2.                      rt_size_t   msg_size,
3.                      rt_size_t   max_msgs,
4.                      rt_uint8_t  flag)

复制代码

• 静态创建消息队列
  

1. rt_err_t rt_mq_init(rt_mq_t     mq,
2.                     const char *name,
3.                     void       *msgpool,
4.                     rt_size_t   msg_size,
5.                     rt_size_t   pool_size,
6.                     rt_uint8_t  flag)

复制代码

2、消息的发送

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

• 直接发送消息
  

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

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

复制代码

• 等待发送消息
  

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

1. rt_err_t rt_mq_send_wait(rt_mq_t     mq,
2.                          const void *buffer,
3.                          rt_size_t   size,
4.                          rt_int32_t  timeout)

复制代码

• 紧急发送消息
  

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

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

复制代码

3、消息的接收

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

1. rt_err_t rt_mq_recv(rt_mq_t    mq,
2.                     void      *buffer,
3.                     rt_size_t  size,
4.                     rt_int32_t timeout)

复制代码

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

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

1. rt_err_t rt_mq_delete(rt_mq_t mq)

复制代码

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

1. rt_err_t rt_mq_detach(rt_mq_t mq)

复制代码

二、创建消息队列示例

1、创建消息队列

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

复制代码

2、完整示例

1. #include "thread_task.h"
2. #include "main.h"
3. #include <stdio.h>      
4. #include "rtthread.h"
5. #include <rthw.h>
7. /******************************************** 线程 1 ******************************************************/
8. #define THREAD_1_PRIORITY                  4           /* 进程优先级 */
9. #define THREAD_1_STACK_SIZE                512         /* 进程栈空间大小 */
10. #define THREAD_1_TIMESLICE                10           /* 进程执行时间片个数 */
11. static struct rt_thread *thread_1_handle;        /* 进程句柄 */
13. /******************************************** 线程 2 ******************************************************/
14. #define THREAD_2_PRIORITY                  5           /* 进程优先级 */
15. #define THREAD_2_STACK_SIZE                512         /* 进程栈空间大小 */
16. #define THREAD_2_TIMESLICE                10           /* 进程执行时间片个数 */
17. static struct rt_thread *thread_2_handle;    /* 进程句柄 */
19. /* 消息队列 */
20. rt_mq_t mq_test;
22. /**
23. * @brief   线程1入口函数
24. * @param   无
25. * @retval  无
26. */
27. void thread_1_entry(void* param)
28. {
29.     uint8_t buffer[2];
30.     while(1)
31.     {
33.         rt_mq_recv(mq_test, buffer, 2, 0xffffffff);
35.         HAL_GPIO_TogglePin(GPIOC, LED1_Pin);
37.         rt_thread_delay(4000);     /* 精准延时1000时间片 */
38.     }
39. }
41. /**
42. * @brief   线程2入口函数
43. * @param   无
44. * @retval  无
45. */
46. void thread_2_entry(void* param)
47. {
48.     uint8_t text[2] = {0,1};
49.     while(1)
50.     {
51.         text[0] ++;
52.         rt_mq_send_wait(mq_test, text, 2, 0xffff);
54.         HAL_GPIO_TogglePin(GPIOC, LED2_Pin);
55.         
56.         rt_thread_delay(200);   
57.     }
58. }
62. /**
63. * @brief   动态创建线程任务并启动
64. * @param   无
65. * @retval  无
66. */
67. void ThreadStart(void)
68. {
69.     rt_base_t level = rt_hw_interrupt_disable();
70.     /* 动态创建线程 */
71.     thread_1_handle = rt_thread_create(
72.         "thread_1",                                /* 线程句柄名称*/
73.         thread_1_entry,                        /* 函数入口 */
74.         RT_NULL,                                /* 入口函数参数 */
75.         THREAD_1_STACK_SIZE,        /* 线程栈大小 */
76.         THREAD_1_PRIORITY,                /* 线程优先级 */
77.         THREAD_1_TIMESLICE          /* 线程时间片大小 */
78.     );
80.                
81.         rt_thread_startup(thread_1_handle); /* 启动线程 */
83.     thread_2_handle = rt_thread_create(
84.             "thread_2",                                /* 线程句柄名称*/
85.             thread_2_entry,                        /* 函数入口 */
86.             RT_NULL,                                /* 入口函数参数 */
87.             THREAD_2_STACK_SIZE,        /* 线程栈大小 */
88.             THREAD_2_PRIORITY,                /* 线程优先级 */
89.             THREAD_2_TIMESLICE          /* 线程时间片大小 */
90.         );
91.                
92.         rt_thread_startup(thread_2_handle); /* 启动线程 */
94.         /* 动态创建一个消息队列 */
95.     mq_test = rt_mq_create(
96.         "mq_test",              /* 消息队列名称 */
97.         2,                      /* 消息大小 */
98.         12,                     /* 消息数量 */
99.         RT_IPC_FLAG_FIFO        /* 选择线程唤醒的顺序方式 */
100.     );
101.     rt_hw_interrupt_enable(level);
102. }

复制代码

执行流程如下:

• 初始阶段
  

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

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

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！