使用消息队列做同步应用

发布于 2019-02-13 19:07:37 浏览：3060 订阅该版

**同步消息**在一般的系统设计中会经常遇到要发送同步消息的问题，这个时候就可以根据当时状态的不同选择相应的实现：两个线程间可以采用**[消息队列 + 信号量或邮箱]**的形式实现。

发送线程通过消息发送的形式发送相应的消息给消息队列，发送完毕后希望获得接收线程的收到确认，
工作示意图如下图所示：
![07msg_syn.png](https://www.rt-thread.org/document/site/programming-manual/ipc2/figures/07msg_syn.png)

看到消息队列中有提到这个，就想简单试一下：
功能改造原来的msgq_sample 中 5ms 发送20个数据到消息队列，然后50ms 以后接收。
改造成：只有收到确认信息，才发送。这里使用邮箱来确认。
代码如下：主要就是做个记录：
```c
/*
 * Copyright (c) 2006-2018, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2018-08-24     yangjie      the first version
 */

/*
 * 程序清单：消息队列例程
 *
 * 这个程序会创建2个动态线程，一个线程会从消息队列中收取消息；一个线程会定时给消
 * 息队列发送 普通消息和紧急消息。
 */
#include <rtthread.h>

#define THREAD_PRIORITY      25
#define THREAD_TIMESLICE     5

/* 消息队列控制块 */
static struct rt_messagequeue mq;
/* 邮箱控制块 */
static struct rt_mailbox mb;
/* 消息队列中用到的放置消息的内存池 */
static rt_uint8_t msg_pool[2048];
/* 用于放邮件的内存池 */
static char mb_pool[128];
static char mb_str1[] = "I'm a mail!";
ALIGN(RT_ALIGN_SIZE)
static char thread1_stack[1024];
static struct rt_thread thread1;
struct msg
{
    rt_uint8_t *data_ptr;    /* 数据块首地址 */
    rt_uint32_t data_size;   /* 数据块大小   */
		struct rt_mailbox ack;
		
};
/* 线程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)
 {
					 /* 发送 mb_str1 地址到邮箱中 */
 rt_mb_send(&mb, (rt_uint32_t)&mb_str1);
 rt_kprintf("thread1: recv msg from msg queue, the content:%c ", buf);
 if (cnt == 19)
 {
 break;
 }
 }
 /* 延时50ms */
 cnt++;
 rt_thread_mdelay(50);
 }
 rt_kprintf("thread1: detach mq ");
 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;
		char *str;
 while (1)
 {
 if (cnt == 8)
 {
 /* 发送紧急消息到消息队列中 */
 result = rt_mq_urgent(&mq, &buf, 1);
 if (result != RT_EOK)
 {
 rt_kprintf("rt_mq_urgent ERR ");
 }
 else
 {
 rt_kprintf("thread2: send urgent message - %c ", buf);
 }
 }
 else if (cnt >= 20)/* 发送20次消息之后退出 */
 {
 rt_kprintf("message queue stop send, thread2 quit ");
 break;
 }
 else
 {
				rt_kprintf("thread2: try to recv a mail ");

/* 从邮箱中收取邮件 */
 if ((cnt == 0) || (rt_mb_recv(&mb, (rt_uint32_t *)&str, RT_WAITING_FOREVER) == RT_EOK))
				{
						rt_kprintf("thread2: get a mail from mailbox, the content:%s ", str);
 /* 发送消息到消息队列中 */
 result = rt_mq_send(&mq, &buf, 1);
 if (result != RT_EOK)
 {
 rt_kprintf("rt_mq_send ERR ");
 }

rt_kprintf("thread2: send message - %c ", 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. ");
 return -1;
 }
		
 /* 初始化一个 mailbox */
 result = rt_mb_init(&mb,
 "mbt", /* 名称是 mbt */
 &mb_pool[0], /* 邮箱用到的内存池是 mb_pool */
 sizeof(mb_pool) / 4, /* 邮箱中的邮件数目，因为一封邮件占 4 字节 */
 RT_IPC_FLAG_FIFO); /* 采用 FIFO 方式进行线程等待 */
 if (result != RT_EOK)
 {
 rt_kprintf("init mailbox failed. ");
 return -1;
 }
 rt_thread_init(&thread1,
 "thread1",
 thread1_entry,
 RT_NULL,
 &thread1_stack[0],
 sizeof(thread1_stack),
 THREAD_PRIORITY, THREAD_TIMESLICE);
 rt_thread_startup(&thread1);

rt_thread_init(&thread2,
                   "thread2",
                   thread2_entry,
                   RT_NULL,
                   &thread2_stack[0],
                   sizeof(thread2_stack),
                   THREAD_PRIORITY, THREAD_TIMESLICE);
    rt_thread_startup(&thread2);

return 0;
}

/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(msgq_sample, msgq sample);
```
![55.png](/uploads/201902/13/190729d78r8gcxzej4cdav.png)