目录

消息队列的基本概念

消息队列的运作机制

消息队列常用接口

消息队列控制块

删除消息队列

初始化消息队列

脱离消息队列

发送消息

发送紧急消息

接收消息

使用场合

示例代码

消息队列的基本概念

消息队列是一种常用的线程间通讯方式，它能够接收来自线程或中断服务例程中不固定长度的消息，并把消息缓存在自己的内存空间中，其他线程也能够从消息队列中读取相应的消息，而当消息队列是空的时候，就可以挂起读取线程。当有新的消息到达时，挂起的线程将被唤醒以接收并处理消息。消息队列是一种异步通信方式。

RT-Thread中使用消息队列数据结构实现线程异步通信工作，具有如下特性：

• 消息支持先进先出方式排队和优先级排队方式，支持异步读写工作方式
• 读队列支持超时机制
• 支持发送紧急消息，这里的紧急消息是往队列头发送消息
• 可以允许不同长度（不超过节点最大值）的任意类型消息
• 一个线程能够从任意一个消息队列接收和发送消息
• 多个线程能够从同一个消息队列接收和发送消息
• 当队列使用结束后，需要通过删除队列操作，释放内存回收

如下图所示，消息队列的工作示意图，通过消息队列服务，线程或中断服务例程可以将一条或者多条消息放入消息队列中，同样一个或者多个线程可以从消息队列中获取消息。当有多个消息发送到消息队列中时，通常应该将先进入消息队列的消息先传给线程，也就是说，线程先得到的是最先进入消息队列的消息，即先进先出原则（FIFO）。

 消息队列的本质是链表：其中核心是链表和定时器。

RT-Thread操作系统的消息队列对象由多个元素组成，当消息队列被创建时，它就被分配到了消息队列控制块；消息队列名称，内存缓冲区，消息大小以及队列长度等，同时每个消息队列对象中包含着 多个消息框，每个消息框可以存放一条消息，消息队列中的第一个和最后一个消息框被分别称为消息链表头和消息链表尾，对应于消息队列控制块中的msg_queue_head和msg_queue_tail；有些消息框可能是空的，他们通过msg_queue_free形成一个空闲消息框链表。所有消息队列中的消息框总数即是消息队列的长度。这个长度可以在创建消息队列时指定。

消息队列的运作机制

只要知道消息队列的句柄，谁都可以读，写该消息队列

线程、ISR都可读、写消息队列。可以多个线程读写消息队列

线程读写消息队列时，如果读写不成功，可以即刻返回错误，也可以阻塞，阻塞时可指定超时时间，口语化的说，就是可以定个闹钟，如果能读写了就马上进入就绪态，否则就阻塞直到超时。

比如，某个线程读消息队列时：

• 如果消息队列中有可用的 消息，即可得到消息并返回
• 如果消息队列中没有可用的消息，线程有两种选择，即可放回一个错误值，或者阻塞一段时间
• 如果线程阻塞，它何时被唤醒？
  • 在指定的时间内，别的线程或者中断服务程序写了队列，会把它唤醒
  • 否则，指定的时间到达后超时返回错误

既然读写消息队列的线程个数没有限制：

• 多个线程都想写队列，但是队列已经满了，这些线程可以进入阻塞态；它们都在等待队列有空间
• 那么队列有空间时，会把哪个线程唤醒？
• 多个线程都想读队列，但是队列已经空了，这些线程可以进入阻塞态；它们都在等待队列有数据
• 那么：队列有数据，把那个线程先唤醒？
• 唤醒谁？有两种方法，创建队列时，可以在指定一个参数flag
• RT_IPC_FLAG_FIFO:表示唤醒等待时间最长的线程
• RT_IPC_FLAG_PRIO：表示唤醒优先级最高的等待线程

下面以一个具体的消息队列的简化操作做一个概述

消息队列的简化操作如下图所示，从图中可知

• 消息队列可以包含若干个消息
• 每个消息可以容纳的数据大小是一样的
• 创建消息队列时就要指定长度（消息个数）、消息的数据大小
• 数据的操作采用先进先出的方法，写数据时放到尾部，读数据时从头部开始读
• 也可以把紧急的数据写到消息队列的头部

 详细过程如下

传输数据的两种方法

使用消息队列传输数据时有两种方法：

• 拷贝：把数据、把变量的值复制进消息队列里
• 引用：把数据、把变量的地址复制进消息队列里

RT-Thread使用拷贝值（memcpy）的方法，这更简单：

•  局部变量的值可以发送到消息队列中，后续即使函数退出，局部变量被收回，也不会影响消息队列中的数据
• 无需分配buffer来保存数据，消息队列中有buffer
• 局部变量可以马上再次使用
• 发送线程，接收线程解耦，接收线程不需要知道这数据是谁的，也不需要发送线程来释放数据
• 如果数据实在太大，你可以使用消息队列传输它的地址
• 消息队列的空间有RT-Thread内核分配，无需线程操心

消息队列常用接口

消息队列控制块

struct rt_messagequeue
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    void                *msg_pool;                      /**< start address of message queue */

    rt_uint16_t          msg_size;                      /**< message size of each message */
    rt_uint16_t          max_msgs;                      /**< max number of messages */

    rt_uint16_t          entry;                         /**< index of messages in the queue */

    void                *msg_queue_head;                /**< list head */
    void                *msg_queue_tail;                /**< list tail */
    void                *msg_queue_free;                /**< pointer indicated the free node of queue */

    rt_list_t            suspend_sender_thread;         /**< sender thread suspended on this message queue */
};
typedef struct rt_messagequeue *rt_mq_t;

struct rt_ipc_object
{
    struct rt_object parent;                            /**< inherit from rt_object */

    rt_list_t        suspend_thread;                    /**< threads pended on this resource */
};

其中不难看到，消息队列属于内核对象的一种，从属于IPC对象管理器，控制块中有IPC对象结构体，里面有一个等待读的阻塞列表，外加一个发送阻塞的链表，所以说链表是消息队列的核心，消息队列的本质是链表。

创建消息队列:消息队列在使用前，应该被创建出来，或对已有的静态消息队列对象进行初始化，创建消息队列的函数接口如下所示：

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

 创建消息队列时先创建一个消息队列对象控制块，然后给消息队列分配一个块内存空间，组织成空闲消息链表，这块内存的大小等于[消息大小+消息头（用于链表连接）]与雄安锡队列容量的乘积，接着再初始化消息队列，此时消息队列为空。

参数	描述
name msg_size max_msgs flag	消息队列的名称 消息队列中一条消息的最大长度 消息队列的最大容量 消息队列采用的等待方式，可以取值

删除消息队列

当消息队列不再使用时，应该删除它以释放系统资源，一旦操作完成，消息队列将被永久性的删除。删除消息队列的函数接口如下：

rt_err_t rt_mq_delete(rt_mq_t mq)

删除消息队列时，如果有线程被挂起在该消息队列的等待队列上，则内核先唤醒挂起在该消息等待队列上的所有线程（返回值是-RT_ERROR），然后再释放消息队列使用的内存，最后删除消息队列对象。

参数	描述
mq	消息队列对象的句柄

初始化消息队列

初始化静态消息队列对象跟创建消息队列对象类似，只是静态消息队列对象的内存是在系统编译时由编译器分配的，一般存放于数据段或ZI段中。在使用这类静态消息队列对象前，需要进行初始化。初始化消息队列对象的函数接口如下：

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)

参数	描述
mq name msgpool msg_size pool_size flag	指向静态消息队列对象的句柄 消息队列的名称 用于存放消息的缓冲区 消息队列中一条消息的最大长度 存放消息的缓冲区的大小 消息队列采用的等待方式，可以取值

脱离消息队列

脱离消息队列将使消息队列对象从内核对象管理器中删除，脱离消息队列使用下面的接口

rt_err_t rt_mq_detach(rt_mq_t mq)

使用该函数接口后，内核先唤醒所有挂在该消息等待队列对象上的线程（返回值是-RT_ERROR），然后将该消息队列对象从内核对象管理器中删除

参数	描述
mq	指向静态消息队列对象的句柄

发送消息

线程或者中断服务程序都可以给消息队列发送消息。当发送消息时，消息队列对象先从空闲消息链表上取下一个空闲消息块，把线程或者中断服务程序发送的消息内容复制到消息块上，然后把该消息块挂到消息队列的尾部。当且仅当空闲消息链表上有可用的空闲消息块时，发送者才能成功发送消息；当空闲消息链表上无可用消息块，说明消息队列已满，此时发送消息的线程或者中断程序会收到一个错误码（-RT_EFULL）。发送消息的函数接口如下：

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

参数	描述
mq buffer size timeout	消息队列对象的句柄 消息内容 消息大小 超时时间

发送紧急消息

发送紧急消息的过程与发送消息几乎一样，唯一不同的是，当发送紧急消息时，从空闲消息链表上取下来的消息块不是挂到消息队列的队尾，而是挂到队首，这样接收者就能够优先接收到紧急消息，从而及时的进行消息处理。发送紧急消息的函数接口如下：

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

参数	描述
mq buffer size	消息队列对象的句柄 消息内容 消息大小

接收消息

当消息队列中有消息时，接收者才能接收消息，否则接收者会根据超时时间设置或挂起在消息队列的等待线程上，或直接返回。接收消息函数接口如下：

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

接收消息时，接收者需要指定存储消息的消息队列对象句柄，并且指定一个内存缓冲区，接收到的消息内容将被复制到该缓冲区。此外，还需指定未能及时取到消息时的超时时间。接收到一个消息后，消息队列上的队首消息被转移到了空闲消息链表的尾部。

参数	描述
mq buffer size timeout	消息队列对象的句柄 用于接收消息的数据块 消息大小 指定的超时时间

使用场合

消息队列可以应用于发送不定长消息的场合，包括线程与线程间的消息交换，以及中断服务例程中发送给线程的消息（中断服务例程不可能接收消息）

典型使用

消息队列和邮箱的明显不同是消息的长度并不限定在4字节以内，另外消息队列也包括了发送一个发送紧急消息的函数接口。但是当创建的是一个所有消息的最大长度是4字节的消息队列时，消息队列对象将蜕化从邮箱。这个不限定长度的消息，也及时的反应到了代码上。

示例代码

/* 
 * 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>

/* 消息队列控制块 */
static struct rt_messagequeue mq;
/* 消息队列中用到的放置消息的内存池 */
static rt_uint8_t msg_pool[2048];

typedef enum{
	Thread2,
	Thread3
}ID_t;

typedef struct
{
	ID_t DataID;
	char data;
}Data_t;

ALIGN(RT_ALIGN_SIZE)
static char thread1_stack[1024];
static struct rt_thread thread1;
/* 线程1入口函数 */
static void thread1_entry(void *parameter)
{
	Data_t buf;
    rt_uint8_t cnt = 0;

    while (1)
    {
        /* 从消息队列中接收消息 */
        if (rt_mq_recv(&mq, &buf, sizeof(buf), RT_WAITING_FOREVER) == RT_EOK)
        {
			if(buf.DataID == Thread2)
			{
				rt_kprintf("thread1: recv msg from Thread2 msg queue, the content:%c\n", buf.data);

			}
			else if(buf.DataID == Thread3 )
			{
				rt_kprintf("thread1: recv msg from Thread3 msg queue, the content:%c\n",buf.data);
			
			}
			
            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;
    Data_t buff = {Thread2,'A'};    
    rt_uint8_t cnt = 0;
    
    while (1)
    {
        if (cnt == 8)
        {
            /* 发送紧急消息到消息队列中 */
            result = rt_mq_urgent(&mq, &buff, sizeof(Data_t));
            if (result != RT_EOK)
            {
                rt_kprintf("rt_mq_urgent ERR\n");
            }
            else
            {
                rt_kprintf("thread2: send urgent message - %c\n", buff.data);
            }
        }
        else if (cnt >= 20)/* 发送20次消息之后退出 */
        {
            rt_kprintf("message queue stop send, thread2 quit\n");
            break;
        }
        else
        {
            /* 发送消息到消息队列中 */
            result = rt_mq_send(&mq, &buff, sizeof(buff));
            if (result != RT_EOK)
            {
                rt_kprintf("rt_mq_send ERR\n");
            }

            rt_kprintf("thread2: send message - %c\n", buff.data);
        }
        buff.data++;
        cnt++;
        /* 延时5ms */
        rt_thread_mdelay(10);
    }
}

ALIGN(RT_ALIGN_SIZE)
static char thread3_stack[1024];
static struct rt_thread thread3;

void thread3_entry(void *parameter)
{
	int result;
    Data_t buff = {Thread3,'a'};    
    rt_uint8_t cnt = 0;
    
    while (1)
    {
        if (cnt == 5)
        {
            /* 发送紧急消息到消息队列中 */
            result = rt_mq_urgent(&mq, &buff, sizeof(Data_t));
            if (result != RT_EOK)
            {
                rt_kprintf("rt_mq_urgent ERR\n");
            }
            else
            {
                rt_kprintf("thread3: send urgent message - %c\n", buff.data);
            }
        }
        else if (cnt >= 20)/* 发送20次消息之后退出 */
        {
            rt_kprintf("message queue stop send, thread3 quit\n");
            break;
        }
        else
        {
            /* 发送消息到消息队列中 */
            result = rt_mq_send(&mq, &buff, sizeof(buff));
            if (result != RT_EOK)
            {
                rt_kprintf("rt_mq_send ERR\n");
            }

            rt_kprintf("thread3: send message - %c\n", buff.data);
        }
        buff.data++;
        cnt++;
        /* 延时5ms */
        rt_thread_mdelay(10);
    }
		

}

/* 消息队列示例的初始化 */
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);
	
	rt_thread_init(&thread3,
					"thread3",
					thread3_entry,
					RT_NULL,
					&thread3_stack[0],
					sizeof(thread3_stack),
					25,
					5
					);
					
	rt_thread_startup(&thread3);

    return 0;
}

/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(msgq_sample, msgq sample);


/*
	个人总结
	·消息队列：消息队列是一种常用的线程间的通讯方式，能够接收来自线程或者中断服务例程中不定长的消息，
	并把消息缓存在自己的内存空间中，是一种异步通信方式。
	·消息队列控制块
	
		struct rt_messagequeue
		{
			struct rt_ipc_object parent;                        **< IPC对象结构体 

			void                *msg_pool;                      **< 存放消息的消息池开始地址

			rt_uint16_t          msg_size;                      **< 每个消息的长度
			rt_uint16_t          max_msgs;                      **< 最大能够容纳的消息数

			rt_uint16_t          entry;                         **< 队列中已有的消息数*

			void                *msg_queue_head;                **< 消息链表头 *
			void                *msg_queue_tail;                **< 消息链表尾*
			void                *msg_queue_free;                *< 空闲消息链表 *

			rt_list_t            suspend_sender_thread;         **<发送等待列表
		};
	
	·消息队列相关接口
		创建消息队列：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)
					 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_delete(rt_mq_t mq)
					  rt_err_t rt_mq_detach(rt_mq_t mq)
		发送消息
				rt_err_t rt_mb_send(rt_mailbox_t mb, rt_ubase_t value)

		接收消息
				rt_err_t rt_mq_recv(rt_mq_t    mq,
									void      *buffer,
									rt_size_t  size,
									rt_int32_t timeout)
									
		对于多个线程向同一个消息队列发送消息，可以构造消息类型，以便其他线程在接收消息时能够加以区分，是哪个线程的相应的消息。
		
	重点掌握：
		1、消息队列的原理和适用场合
		2、消息的各种API函数接口
		3、消息队列的内部机制（链表+定时器）
			读消息队列：
				1、不等:直接返回错误
				2、等：等多久还是一直等，等待的过程中把自己阻塞（从就绪链表中移除）以及把自己加入消息队列的读阻塞列表中（以便
				其他线程写了消息队列去唤醒）
				3、再次运行：包括两种情况：超时唤醒和被其他线程写消息队列唤醒
				

*/