Rt-Thread学习笔记-----消息队列（四）
消息队列
消息队列是另一种常用的线程间通讯方式，是邮箱的扩展。可以应用在多种场合：线程间的消息交换、使用串口接收不定长数据等。
消息队列的工作机制
消息队列能够接收来自线程或中断服务例程中不固定长度的消息，并把消息缓存在自己的内存空间中。其他线程也能够从消息队列中读取相应的消息，而当消息队列是空的时候，可以挂起读取线程。当有新的消息到达时，挂起的线程将被唤醒以接收并处理消息。消息队列是一种异步的通信方式。
如下图所示，线程或中断服务例程可以将一条或多条消息放入消息队列中。同样，一个或多个线程也可以从消息队列中获得消息。当有多个消息发送到消息队列时，通常将先进入消息队列的消息先传给线程，也就是说，线程先得到的是最先进入消息队列的消息，即先进先出原则 (FIFO)。
RT-Thread 操作系统的消息队列对象由多个元素组成，当消息队列被创建时，它就被分配了消息队列控制块：消息队列名称、内存缓冲区、消息大小以及队列长度等。同时每个消息队列对象中包含着多个消息框，每个消息框可以存放一条消息；消息队列中的第一个和最后一个消息框被分别称为消息链表头和消息链表尾，对应于消息队列控制块中的 msg_queue_head 和 msg_queue_tail；有些消息框可能是空的，它们通过 msg_queue_free 形成一个空闲消息框链表。所有消息队列中的消息框总数即是消息队列的长度，这个长度可在消息队列创建时指定。
常用消息队列的函数
创建消息队列 rt_mq_create。
写队列操作函数 rt_mq_send。
读队列操作函数 rt_mq_recv。
删除队列 rt_mq_delete。
初始化和脱离消息队列 rt_mq_init。rt_mq_detach
发送紧急消息 rt_mq_urgent。
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);
//函数举例
/* 创建一个消息队列 */
test_mq = rt_mq_create("test_mq", /* 消息队列名字 */
40, /* 消息的最大长度 */
20, /* 消息队列的最大容量 */
RT_IPC_FLAG_FIFO);/* 队列模式 FIFO(0x00)*/
if (test_mq != RT_NULL)
rt_kprintf("消息队列创建成功！\n\n");
2、消息队列删除函数：rt_mq_delete
1、当消息队列不再被使用时，应该删除它以释放系统资源，一旦操作完成，消息队列将被永久性地删除。
2、如果某个消息队列没有被创建，那也是无法被删除的。
//函数原型
rt_err_t rt_mq_delete(rt_mq_t mq);
//举例
/* 定义消息队列控制块 */
static rt_mq_t test_mq = RT_NULL;
rt_err_t uwRet = RT_EOK;
uwRet = rt_mq_delete(test_mq);
if (RT_EOK == uwRet)
rt_kprintf(" 消息队列删除成功！\n\n");
删除消息队列时，如果有线程被挂起在该消息队列等待队列上，则内核先唤醒挂起在该消息等待队列上的所有线程（线程返回值是 - RT_ERROR），然后再释放消息队列使用的内存，最后删除消息队列对象。
3、消息队列发送消息函数：rt_mq_send
线程或者中断服务程序都可以给消息队列发送消息。当发送消息时，消息队列对象先从空闲消息链表上取下一个空闲消息块，把线程或者中断服务程序发送的消息内容复制到消息块上，然后把该消息块挂到消息队列的尾部。当且仅当空闲消息链表上有可用的空闲消息块时，发送者才能成功发送消息；当空闲消息链表上无可用消息块，说明消息队列已满，此时，发送消息的的线程或者中断程序会收到一个错误码（-RT_EFULL）。
//原型
rt_err_t rt_mq_send (rt_mq_t mq, void* buffer, rt_size_t size);
//举例
static void send_thread_entry(void* parameter)
{
rt_err_t uwRet = RT_EOK;
uint32_t send_data1 = 1;
uint32_t send_data2 = 2;
while (1) {/* K1 被按下 */
if ( Key_Scan(KEY1_GPIO_PORT,KEY1_GPIO_PIN) == KEY_ON ) {
/* 将数据写入（发送）到队列中，等待时间为 0 */
uwRet = rt_mq_send(test_mq, /* 写入（发送）队列的 ID( 句柄) */
&send_data1, /* 写入（发送）的数据 */
sizeof(send_data1)); /* 数据的长度 */
if (RT_EOK != uwRet) {
rt_kprintf("数据不能发送到消息队列！错误代码: %lx\n",uwRet);
}
}/* K1 被按下 */
if ( Key_Scan(KEY2_GPIO_PORT,KEY2_GPIO_PIN) == KEY_ON ) {
/* 将数据写入（发送）到队列中，等待时间为 0 */
uwRet = rt_mq_send( test_mq, /* 写入（发送）队列的 ID( 句柄) */
&send_data2, /* 写入（发送）的数据 */
sizeof(send_data2)); /* 数据的长度 */
if (RT_EOK != uwRet) {
rt_kprintf("数据不能发送到消息队列！错误代码: %lx\n",uwRet);
}
}/* K2 被按下 */
rt_thread_delay(20);
}
}
4、消息队列接收消息函数：rt_mq_recv
当消息队列中有消息时，接收者才能接收消息，否则接收者会根据超时时间设置，或挂起在消息队列的等待线程队列上，或直接返回。
//接口函数原型
rt_err_t rt_mq_recv (rt_mq_t mq, void* buffer,
rt_size_t size, rt_int32_t timeout);
//举例
/* 队列读取（接收），等待时间为一直等待 */
uwRet = rt_mq_recv(test_mq, /* 读取（接收）队列的 ID( 句柄) */
&r_queue, /* 读取（接收）的数据保存位置 */
sizeof(r_queue), /* 读取（接收）的数据的长度 */
RT_WAITING_FOREVER); /* 等待时间：一直等 */
if (RT_EOK == uwRet){
rt_kprintf("本次接收到的数据是：%d\n",r_queue);
} else{
rt_kprintf("数据接收出错,错误代码: 0x%lx\n",uwRet);
}
接收消息时，接收者需指定存储消息的消息队列对象句柄，并且指定一个内存缓冲区，接收到的消息内容将被复制到该缓冲区里。此外，还需指定未能及时取到消息时的超时时间。如下图所示，接收一个消息后消息队列上的队首消息被转移到了空闲消息链表的尾部。
消息队列使用注意事项：
在使用 RT-Thread提供的消息队列函数的时候，需要了解以下几点：
1. 使用 rt_mq_recv()、rt_mq_send()、rt_mq_delete()等这些函数之前应先创建需消息队列，并根据队列句柄进行操作。
2. 队列读取采用的是先进先出（FIFO）模式，会首先读取出首先存储在队列中的数据。当然也有例外，RT-Thread 给我们提供了另一个函数，可以发送紧急消息的，那么读取的时候就会读取到紧急消息的数据。
3. 必须要我们定义一个存储读取出来的数据的地方，并且把存储数据的起始地址传递给 rt_mq_recv()函数，否则，将发生地址非法的错误。
4. 接收消息队列中的消息是拷贝的方式，读取消息时候定义的地址必须保证能存放下即将读取消息的大小。
5、初始化和脱离消息队列：rt_mq_init 、rt_mq_detach
5.1 初始化静态消息队列对象跟创建消息队列对象类似，只是静态消息队列对象的内存是在系统编译时由编译器分配的，一般放于读数据段或未初始化数据段中。在使用这类静态消息队列对象前，需要进行初始化。
//函数接口
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);
区指针、消息大小以及消息队列缓冲区大小。如下图所示，消息队列初始化后所有消息都挂在空闲消息链表上，消息队列为空。下表描述了该函数的输入参数与返回值：
5.2 脱离消息队列将使消息队列对象被从内核对象管理器中脱离。
rt_err_t rt_mq_detach(rt_mq_t mq);
使用该函数接口后，内核先唤醒所有挂在该消息等待队列对象上的线程（线程返回值是 -RT_ERROR），然后将该消息队列对象从内核对象管理器中脱离。下表描述了该函数的输入参数与返回值：
6、发送紧急消息 ：rt_mq_urgent
发送紧急消息的过程与发送消息几乎一样，唯一的不同是，当发送紧急消息时，从空闲消息链表上取下来的消息块不是挂到消息队列的队尾，而是挂到队首，这样，接收者就能够优先接收到紧急消息，从而及时进行消息处理。发送紧急消息的函数接口如下：
rt_err_t rt_mq_urgent(rt_mq_t mq, void* buffer, rt_size_t size);
对应静态创建消息队列例程：
这是一个消息队列的应用例程，例程中初始化了 2 个静态线程，一个线程会从消息队列中收取消息；另一个线程会定时给消息队列发送普通消息和紧急消息，如下代码所示：
/* 消息队列控制块 */
static struct rt_messagequeue mq;
/* 消息队列中用到的放置消息的内存池 */
static rt_uint8_t msg_pool[2048];
ALIGN(RT_ALIGN_SIZE)
static char thread1_stack[1024];
static struct rt_thread thread1;
/* 线程 1 入口函数 */
static void thread1_entry(void *parameter)
{
char buf = 0;
rt_uint8_t cnt = 0;
while (1)
{
/* 从消息队列中接收消息 */
if (rt_mq_recv(&mq, &buf, sizeof(buf), RT_WAITING_FOREVER) == RT_EOK)
{
rt_kprintf("thread1: recv msg from msg queue, the content:%c\n", buf);
if (cnt == 19)
{
break;
}
}
/* 延时 50ms */
cnt++;
rt_thread_mdelay(50);
}
rt_kprintf("thread1: detach mq \n");
rt_mq_detach(&mq);//脱离消息队列 该函数将使消息队列对象被从内核对象管理器中删除
}
ALIGN(RT_ALIGN_SIZE)
static char thread2_stack[1024];
static struct rt_thread thread2;
/* 线程 2 入口 */
static void thread2_entry(void *parameter)
{
int result;
char buf = 'A';
rt_uint8_t cnt = 0;
while (1)
{
if (cnt == 8)
{
/* 发送紧急消息到消息队列中 */
result = rt_mq_urgent(&mq, &buf, 1);
if (result != RT_EOK)
{
rt_kprintf("rt_mq_urgent ERR\n");
}
else
{
rt_kprintf("thread2: send urgent message - %c\n", buf);
}
}
else if (cnt>= 20)/* 发送 20 次消息之后退出 */
{
rt_kprintf("message queue stop send, thread2 quit\n");
break;
}
else
{
/* 发送消息到消息队列中 */
result = rt_mq_send(&mq, &buf, 1);
if (result != RT_EOK)
{
rt_kprintf("rt_mq_send ERR\n");
}
rt_kprintf("thread2: send message - %c\n", buf);
}
buf++;
cnt++;
/* 延时 5ms */
rt_thread_mdelay(5);
}
}
/* 消息队列示例的初始化 */
int msgq_sample(void)
{
rt_err_t result;
/* 初始化消息队列 */
result = rt_mq_init(&mq,
"mqt",
&msg_pool[0], /* 内存池指向 msg_pool */
1, /* 每个消息的大小是 1 字节 */
sizeof(msg_pool), /* 内存池的大小是 msg_pool 的大小 */
RT_IPC_FLAG_FIFO); /* 如果有多个线程等待，按照先来先得到的方法分配消息 */
if (result != RT_EOK)
{
rt_kprintf("init message queue failed.\n");
return -1;
}
rt_thread_init(&thread1,
"thread1",
thread1_entry,
RT_NULL,
&thread1_stack[0],
sizeof(thread1_stack), 25, 5);
rt_thread_startup(&thread1);
rt_thread_init(&thread2,
"thread2",
thread2_entry,
RT_NULL,
&thread2_stack[0],
sizeof(thread2_stack), 25, 5);
rt_thread_startup(&thread2);
return 0;
}
下载验证：
例程演示了消息队列的使用方法。线程 1 会从消息队列中收取消息；线程 2 定时给消息队列发送普通消息和紧急消息。由于线程 2 发送消息 “I” 是紧急消息，会直接插入消息队列的队首，所以线程 1 在接收到消息 “B” 后，接收的是该紧急消息，之后才接收消息“C”。
对应动态创建消息队列例程：
/*
*************************************************************************
* 动态创建消息队列部分
*************************************************************************
*/
/*
*************************************************************************
* 变量
*************************************************************************
*/
/* 定义线程控制块 */
static rt_thread_t receive_thread = RT_NULL;
static rt_thread_t send_thread = RT_NULL;
/* 定义消息队列控制块 */
static rt_mq_t test_mq = RT_NULL;
/*
*************************************************************************
* 函数声明
*************************************************************************
*/
static void receive_thread_entry(void* parameter);
static void send_thread_entry(void* parameter);
/*
*************************************************************************
* 线程定义
*************************************************************************
*/
static void receive_thread_entry(void* parameter)
{
rt_err_t uwRet = RT_EOK;
uint32_t r_queue;
/* 任务都是一个无限循环，不能返回 */
while (1)
{
/* 队列读取（接收），等待时间为一直等待 */
uwRet = rt_mq_recv(test_mq, /* 读取（接收）队列的ID(句柄) */
&r_queue, /* 读取（接收）的数据保存位置 */
sizeof(r_queue), /* 读取（接收）的数据的长度 */
RT_WAITING_FOREVER); /* 等待时间：一直等 */
if(RT_EOK == uwRet)
{
rt_kprintf("本次接收到的数据是：%d\n",r_queue);
}
else
{
rt_kprintf("数据接收出错,错误代码: 0x%lx\n",uwRet);
}
rt_thread_delay(200);
}
}
static void send_thread_entry(void* parameter)
{
rt_err_t uwRet = RT_EOK;
uint32_t send_data1 = 1;
uint32_t send_data2 = 2;
while (1)
{
if( Key_Scan(KEY1_GPIO_PORT,KEY1_GPIO_PIN) == KEY_ON )/* K1 被按下 */
{
/* 将数据写入（发送）到队列中，等待时间为 0 */
uwRet = rt_mq_send( test_mq, /* 写入（发送）队列的ID(句柄) */
&send_data1, /* 写入（发送）的数据 */
sizeof(send_data1)); /* 数据的长度 */
if(RT_EOK != uwRet)
{
rt_kprintf("数据不能发送到消息队列！错误代码: %lx\n",uwRet);
}
}
if( Key_Scan(KEY2_GPIO_PORT,KEY2_GPIO_PIN) == KEY_ON )/* K2 被按下 */
{
/* 将数据写入（发送）到队列中，等待时间为 0 */
uwRet = rt_mq_send( test_mq, /* 写入（发送）队列的ID(句柄) */
&send_data2, /* 写入（发送）的数据 */
sizeof(send_data2)); /* 数据的长度 */
if(RT_EOK != uwRet)
{
rt_kprintf("数据不能发送到消息队列！错误代码: %lx\n",uwRet);
}
}
rt_thread_delay(20);
}
}
/* 消息队列示例的初始化 */
int msgq_sample(void)
{
//动态例程
/*
* 开发板硬件初始化，RTT系统初始化已经在main函数之前完成，
* 即在component.c文件中的rtthread_startup()函数中完成了。
* 所以在main函数中，只需要创建线程和启动线程即可。
*/
rt_kprintf("这是一个STM32开发版-RTT消息队列实验！\n");
rt_kprintf("按下K1或者K2发送队列消息\n");
rt_kprintf("receive任务接收到消息在串口回显\n");
/* 创建一个消息队列 */
test_mq = rt_mq_create("test_mq",/* 消息队列名字 */
40, /* 消息的最大长度 */
20, /* 消息队列的最大容量 */
RT_IPC_FLAG_FIFO);/* 队列模式 FIFO(0x00)*/
if (test_mq != RT_NULL)
rt_kprintf("消息队列创建成功！\n\n");
receive_thread = /* 线程控制块指针 */
rt_thread_create( "receive", /* 线程名字 */
receive_thread_entry, /* 线程入口函数 */
RT_NULL, /* 线程入口函数参数 */
512, /* 线程栈大小 */
3, /* 线程的优先级 */
20); /* 线程时间片 */
/* 启动线程，开启调度 */
if (receive_thread != RT_NULL)
rt_thread_startup(receive_thread);
else
return -1;
send_thread = /* 线程控制块指针 */
rt_thread_create( "send", /* 线程名字 */
send_thread_entry, /* 线程入口函数 */
RT_NULL, /* 线程入口函数参数 */
512, /* 线程栈大小 */
2, /* 线程的优先级 */
20); /* 线程时间片 */
/* 启动线程，开启调度 */
if (send_thread != RT_NULL)
rt_thread_startup(send_thread);
else
return -1;
}
下载验证：
上一篇： mybatis查询语句的背后揭秘