用邮箱实现多事件的单向同步

用邮箱实现多事件的单向同步

概述

信号量通常只能实现单一事件或资源的同步，多个事件的单向同步可以使用队列或事件组。

上节讲述了使用事件组同步多个事件，虽然可以通过事件组传递多个事件，并且接收方可以同步不同的 bits 位知道数据的发送方，但事件组还未能解决下述问题：

1）事件组无法解决事件堆积问题。重复触发的事件无法被计数，可能存在丢失。

2）事件组无法对事件的详细信息进行说明。就好比女朋友通知你去一起吃饭，但告诉你“吃随意”一样，难受！

消息队列，也称邮箱，可以用于解决上述问题，本节重点介绍消息队列。

队列的基本模型

队列的简化模型如入下图所示，从此图可知：

• 队列包含一个数据缓冲区，该缓冲区中可以包含若干块数据，每块数据区域可被称为"条目"(item)。

• 每个条目的最大大小固定。队列可以传输变长的数据，最大大小不能超过条目的最大大小（传递指针也是可以的，但指针指定的内存空间必须保证在使用前有效）。

• 创建队列时就要指定条目个数、每个条目数据的最大大小

• 数据的操作通过读索引、写索引采用先进先出的方法(FIFO，First In First Out)：写数据时放到尾部，读数据时从头部读。

• 也可以强制将数据放到队列头部，则下一个要读的数据就是它。

• 使用队列中的数据后，可以删除该项条目，也可以保留该项条目。

• 允许延时发送。即发送数据时，若队列已经满了（没有多余空间存放数据了），可以通过指定延时时间，延时发送该条信息。

队列的不足：

1）消耗的内存空间当然比信号量、事件组要大一些。

2）具备一定的缓存能力，但若发送者的速率过快，而队列的大小没有足够大，则还是存在溢出风险。

队列的操作

队列的基本操作是：创建、发送（send）、接收(接收)。其中发送可以选择发送到队列后（正常排队），或者发送到队列前（直接成为下一个要被接收的消息）。

其中涉及的 API 主要有以下几个，每个 API 的相关信息都标识在注释中。

/*非0：成功，返回队列句柄，以后使用句柄来操作队列
 *NULL：失败，因为内存不足
*/
QueueHandle_t xQueueCreate( UBaseType_t uxQueueLength, // 队列长度，最多能存放多少个条目(item)
							UBaseType_t uxItemSize ); // 每个条目(item)的大小：以字节为单位

/* pxQueue : 复位哪个队列;
 * 返回值: pdPASS(必定成功)
 */
BaseType_t xQueueReset( QueueHandle_t pxQueue);

/* 等同于xQueueSendToBack
 * 往队列尾部写入数据，如果没有空间，阻塞时间为xTicksToWait
 * 返回值: pdPASS：数据成功写入了队列，errQUEUE_FULL：写入失败，因为队列满了。
 */
BaseType_t xQueueSend(
                                QueueHandle_t    xQueue,// 队列句柄，要写哪个队列
                                const void       *pvItemToQueue, // 数据指针，这个数据的值会被复制进队列，
                                TickType_t       xTicksToWait /* 如果队列满则无法写入新数据，可以让任务进入阻塞状态，xTicksToWait表示阻塞的最大时间(Tick Count)。如果被设为0，无法写入数据时函数会立刻返回；如果被设为portMAX_DELAY，则会一直阻塞直到有空间可写*/
                            );

/* 
 * 往队列尾部写入数据，如果没有空间，阻塞时间为xTicksToWait
 */
BaseType_t xQueueSendToBack(
                                QueueHandle_t    xQueue,
                                const void       *pvItemToQueue,
                                TickType_t       xTicksToWait
                            );

/* 
 * 往队列头部写入数据，如果没有空间，阻塞时间为xTicksToWait
 */
BaseType_t xQueueSendToFront(
                                QueueHandle_t    xQueue,
                                const void       *pvItemToQueue,
                                TickType_t       xTicksToWait
                            );

/* 
 * 仅适用于长度为1的队列，在队列上写入数据。如果队列已满，则覆盖队列中保存的值。
 */
xQueueOverwrite(xQueue, 
				pvItemToQueue)；
				
/* 从队列中接收项目，并从队列中移除对应的条目
 * pdPASS：从队列读出数据入
 * errQUEUE_EMPTY：读取失败，因为队列空了。
 */
BaseType_t xQueueReceive( QueueHandle_t xQueue, // 队列句柄，要读哪个队列
                          void * const pvBuffer, // bufer指针，队列的数据会被复制到这个buffer
                          TickType_t xTicksToWait ); // 果队列空则无法读出数据，可以让任务进入阻塞状态

/* 从队列中接收项目而不从队列中移除对应的条目。
 * pdPASS：从队列读出数据入
 * errQUEUE_EMPTY：读取失败，因为队列空了。
 */
BaseType_t xQueuePeek(QueueHandle_t xQueue, // 队列句柄，要读哪个队列
                      void *const pvBuffer, // bufer指针，队列的数据会被复制到这个buffer
                      TickType_t xTicksToWait); // 果队列空则无法读出数据，可以让任务进入阻塞状态
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

注意：发送到队列的消息是通过拷贝（Copy, 复制）方式实现的，这意味着队列存储的数据是原数据的备份，而不是原数据的引用（局部变量的值可以发送到队列中，后续即使函数退出、局部变量被回收，也不会影响队列中的数据）。

队列的等待-通知机制

此外，队列同样实现了类似前述二值信号量的等待-通知机制（也称为pend-post机制），即无消息时，任务可以暂停运行，进入延时等待消息的状态，有消息时可以立即通知相关的任务唤醒运行。

但是不同于前述信号量的简单同步，队列可能被用于多个任务之间的同步，因此需要考虑以下场景：

1）多个任务等待消息：

某个任务读队列时，如果队列没有数据，则该任务可以进入阻塞状态：还可以指定阻塞的时间。如果队列有数据了，则该阻塞的任务会变为就绪态。如果一直都没有数据，则时间到之后它也会进入就绪态。

既然读取队列的任务个数没有限制，那么当多个任务读取空队列时，这些任务都会进入阻塞状态：有多个任务在等待同一个队列的数据。当队列中有数据时，哪个任务会进入就绪态？

• 优先级最高的任务

• 如果大家的优先级相同，那等待时间最久的任务会进入就绪态

2）多个任务发消息：

跟读队列类似，一个任务要写队列时，如果队列满了，该任务也可以进入阻塞状态：还可以指定阻塞的时间。如果队列有空间了，则该阻塞的任务会变为就绪态。如果一直都没有空间，则时间到之后它也会进入就绪态。

既然写队列的任务个数没有限制，那么当多个任务写"满队列"时，这些任务都会进入阻塞状态：有多个任务在等待同一个队列的空间。当队列中有空间时，哪个任务会进入就绪态？

• 优先级最高的任务

• 如果大家的优先级相同，那等待时间最久的任务会进入就绪态。

需求及功能解析

如前所述，在一些情况下，我们仅仅接收到通知还不够，往往还需要知道下述详细的信息来指导下一步的动作：

1）是谁在发送通知

只是采取信号量往往无法实现。

这种情况的需求：某一个任务等待多个事件中任意一个事件的发生。

2）发送通知的目的或者详情是什么

示例中通过定义一个枚举，表示发送者是谁，同时定义一个动作类型，即加 1，减1。来演示这种情况。

其本质在于，队列不仅可以通知，还可以附带一些数据。

示例解析

示例的 log 给出了，task3 收到 task1\task2 的消息，并打印消息的来源、以及具体指示的操作。

Hello world!
This is esp32 chip with 2 CPU core(s), WiFi/BT/BLE, Minimum free heap size: 294416 bytes
received from flag1, event is Add, count:1
received from task2, event is Sub, count:0
received from task2, event is Sub, count:-1
received from flag1, event is Add, count:0
received from flag1, event is Add, count:1
received from task2, event is Sub, count:0
1
2
3
4
5
6
7
8

讨论

消息队列使用注意事项

在使用FreeRTOS提供的消息队列函数的时候，需要了解以下几点：

• 队列读取采用的是先进先出（FIFO）模式，会先读取先存储在队列中的数据。当然也FreeRTOS也支持后进先出（LIFO）模式，那么读取的时候就会读取到后进队列的数据。

• 在获取（Recv）队列中的消息时候，必须要定义一个存储读取数据的地方，并且该数据区域大小不小于消息大小，否则，很可能引发地址非法的错误。

• 无论是发送或者是接收消息都是以值拷贝的方式进行，如果消息过于庞大，可以将消息的地址（指针）作为消息进行发送、接收；同时要注意该指针指向的内存空间要在接收消息时仍处于有效状态，否则可能出现非法引用内存空间的错误。

• 队列是具有自己独立权限的内核对象，并不属于任何任务。通过队列可以实现多个任务向同一队列写入和读出。但是实际中多是一个任务从队列中读取，较少出现多个任务从一个队列读取的情况。

总结

1）相比信号量、事件组。消息队列(也称为邮箱)可以用于多任务的同步，并提供一定的消息缓存、说明消息详细信息的能力。

2）队列的基本操作是：创建、发送（send）、接收(接收)。其中发送可以选择发送到队列后（正常排队），或者发送到队列前（直接成为下一个要被接收的消息）。

3）队列读取通常采用的是先进先出（FIFO）模式，会先读取先存储在队列中的数据。也支持后进先出（LIFO）模式，那么读取的时候就会读取到后进队列的数据。

4）队列提供的等待-通知机制，在同步多任务出现时，依据任务优先级、等待时间进行裁决分配。

资源链接

1）Learning-FreeRTOS-with-esp32 系列博客介绍
2）对应示例的 code 链接 （点击直达代码仓库）

3）下一篇：使用信号量实现简单双向同步