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

RTOS
3.9k words

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

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

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

一、消息队列API函数

1、消息队列的创建

  • 动态创建消息队列
1
2
3
4
rt_mq_t rt_mq_create(const char *name,
                     rt_size_t   msg_size,
                     rt_size_t   max_msgs,
                     rt_uint8_t  flag)
  • 静态创建消息队列
1
2
3
4
5
6
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、消息的发送

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

  • 直接发送消息

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

1
rt_err_t rt_mq_send(rt_mq_t mq, const void *buffer, rt_size_t size)
  • 等待发送消息

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

1
2
3
4
rt_err_t rt_mq_send_wait(rt_mq_t     mq,
                         const void *buffer,
                         rt_size_t   size,
                         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
2
3
4
rt_err_t rt_mq_recv(rt_mq_t    mq,
                    void      *buffer,
                    rt_size_t  size,
                    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
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/* 使用静态创建（确定内存占用）	*/
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、完整示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
#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次）
  1. 队列填满阶段
    • 线程2发送12条消息后队列满 → rt_mq_send_wait()开始阻塞
    • 系统进入双阻塞状态（线程1和2均挂起）
  2. 恢复同步阶段
    • 线程1的rt_thread_delay(4000)到期后：
      • 消费1条消息 → 释放1个队列空间
      • 唤醒线程2继续发送
    • 稳定状态：每4000 ticks消费1条，线程2每200 ticks补充1条