原文 https://blog.csdn.net/jiejiemcu/article/details/86715766

FreeRTOS訊息佇列
基於 FreeRTOS 的應用程式由一組獨立的任務構成——每個任務都是具有獨立許可權的程式。這些獨立的任務之間的通訊與同步一般都是基於作業系統提供的IPC通訊機制，而FreeRTOS 中所有的通訊與同步機制都是基於佇列實現的。
訊息佇列是一種常用於任務間通訊的資料結構，佇列可以在任務與任務間、中斷和任務間傳送資訊，實現了任務接收來自其他任務或中斷的不固定長度的訊息。任務能夠從佇列裡面讀取訊息，當佇列中的訊息是空時，掛起讀取任務，使用者還可以指定掛起的任務時間；當佇列中有新訊息時，掛起的讀取任務被喚醒並處理新訊息，訊息佇列是一種非同步的通訊方式。

佇列特性
1.資料儲存
佇列可以儲存有限個具有確定長度的資料單元。佇列可以儲存的最大單元數目被稱為佇列的“深度”。在佇列建立時需要設定其深度和每個單元的大小。
通常情況下，佇列被作為 FIFO(先進先出)緩衝區使用，即資料由佇列尾寫入，從佇列首讀出。當然，由佇列首寫入也是可能的。
往佇列寫入資料是通過位元組拷貝把資料複製儲存到佇列中；從佇列讀出資料使得把佇列中的資料拷貝刪除。

2.讀阻塞
當某個任務試圖讀一個佇列時，其可以指定一個阻塞超時時間。在這段時間中，如果佇列為空，該任務將保持阻塞狀態以等待佇列資料有效。當其它任務或中斷服務例程往其等待的佇列中寫入了資料，該任務將自動由阻塞態轉移為就緒態。當等待的時間超過了指定的阻塞時間，即使佇列中尚無有效資料，任務也會自動從阻塞態轉移為就緒態。
由於佇列可以被多個任務讀取，所以對單個佇列而言，也可能有多個任務處於阻塞狀態以等待佇列資料有效。這種情況下，一旦佇列資料有效，只會有一個任務會被解除阻塞，這個任務就是所有等待任務中優先順序最高的任務。而如果所有等待任務的優先順序相同，那麼被解除阻塞的任務將是等待最久的任務。

說些題外話，ucos中是具有廣播訊息的，當有多個任務阻塞在佇列上，當傳送訊息的時候可以選擇廣播訊息，那麼這些阻塞的任務都能被解除阻塞。

3.寫阻塞
與讀阻塞想反，任務也可以在寫佇列時指定一個阻塞超時時間。這個時間是當被寫佇列已滿時，任務進入阻塞態以等待佇列空間有效的最長時間。
由於佇列可以被多個任務寫入，所以對單個佇列而言，也可能有多個任務處於阻塞狀態以等待佇列空間有效。這種情況下，一旦佇列空間有效，只會有一個任務會被解除阻塞，這個任務就是所有等待任務中優先順序最高的任務。而如果所有等待任務的優先順序相同，那麼被解除阻塞的任務將是等待最久的任務。

訊息佇列的工作流程
1.傳送訊息
任務或者中斷服務程式都可以給訊息佇列傳送訊息，當傳送訊息時，如果佇列未滿或者允許覆蓋入隊， FreeRTOS 會將訊息拷貝到訊息佇列隊尾，否則，會根據使用者指定的阻塞超時時間進行阻塞，在這段時間中，如果佇列一直不允許入隊，該任務將保持阻塞狀態以等待佇列允許入隊。當其它任務從其等待的佇列中讀取入了資料（佇列未滿），該任務將自動由阻塞態轉為就緒態。當任務等待的時間超過了指定的阻塞時間，即使佇列中還不允許入隊，任務也會自動從阻塞態轉移為就緒態，此時傳送訊息的任務或者中斷程式會收到一個錯誤碼 errQUEUE_FULL。
傳送緊急訊息的過程與傳送訊息幾乎一樣，唯一的不同是，當傳送緊急訊息時，傳送的位置是訊息佇列隊頭而非隊尾，這樣，接收者就能夠優先接收到緊急訊息，從而及時進行訊息處理。
下面是訊息佇列的傳送API介面，函式中有FromISR則表明在中斷中使用的。

1 /*-----------------------------------------------------------*/
2 BaseType_t xQueueGenericSend( QueueHandle_t xQueue, (1)
3 const void * const pvItemToQueue, (2)
4 TickType_t xTicksToWait, (3)
5 const BaseType_t xCopyPosition ) (4)
6 {
7 BaseType_t xEntryTimeSet = pdFALSE, xYieldRequired;
8 TimeOut_t xTimeOut;
9 Queue_t * const pxQueue = ( Queue_t * ) xQueue;
10
11 /* 已刪除一些斷言操作 */
12
13 for ( ;; ) {
14 taskENTER_CRITICAL(); (5)
15 {
16 /* 佇列未滿 */
17 if ( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength )
18 || ( xCopyPosition == queueOVERWRITE ) ) { (6)
19 traceQUEUE_SEND( pxQueue );
20 xYieldRequired =
21 prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition ); (7)
22
23 /* 已刪除使用佇列集部分程式碼 */
24 /* 如果有任務在等待獲取此訊息佇列 */
25 if ( listLIST_IS_EMPTY(&(pxQueue->xTasksWaitingToReceive))==pdFALSE){ (8)
26 /* 將任務從阻塞中恢復 */
27 if ( xTaskRemoveFromEventList(
28 &( pxQueue->xTasksWaitingToReceive ) )!=pdFALSE) { (9)
29 /* 如果恢復的任務優先順序比當前執行任務優先順序還高，
30 那麼需要進行一次任務切換 */
31 queueYIELD_IF_USING_PREEMPTION(); (10)
32 } else {
33 mtCOVERAGE_TEST_MARKER();
34 }
35 } else if ( xYieldRequired != pdFALSE ) {
36 /* 如果沒有等待的任務，拷貝成功也需要任務切換 */
37 queueYIELD_IF_USING_PREEMPTION(); (11)
38 } else {
39 mtCOVERAGE_TEST_MARKER();
40 }
41
42 taskEXIT_CRITICAL(); (12)
43 return pdPASS;
44 }
45 /* 佇列已滿 */
46 else { (13)
47 if ( xTicksToWait == ( TickType_t ) 0 ) {
48 /* 如果使用者不指定阻塞超時時間，退出 */
49 taskEXIT_CRITICAL(); (14)
50 traceQUEUE_SEND_FAILED( pxQueue );
51 return errQUEUE_FULL;
52 } else if ( xEntryTimeSet == pdFALSE ) {
53 /* 初始化阻塞超時結構體變數，初始化進入
54 阻塞的時間xTickCount和溢位次數xNumOfOverflows */
55 vTaskSetTimeOutState( &xTimeOut ); (15)
56 xEntryTimeSet = pdTRUE;
57 } else {
58 mtCOVERAGE_TEST_MARKER();
59 }
60 }
61 }
62 taskEXIT_CRITICAL(); (16)
63 /* 掛起排程器 */
64 vTaskSuspendAll();
65 /* 佇列上鎖 */
66 prvLockQueue( pxQueue );
67
68 /* 檢查超時時間是否已經過去了 */
69 if (xTaskCheckForTimeOut(&xTimeOut, &xTicksToWait)==pdFALSE){ (17)
70 /* 如果佇列還是滿的 */
71 if ( prvIsQueueFull( pxQueue ) != pdFALSE ) { (18)
72 traceBLOCKING_ON_QUEUE_SEND( pxQueue );
73 /* 將當前任務新增到佇列的等待發送列表中
74 以及阻塞延時列表，延時時間為使用者指定的超時時間xTicksToWait */
75 vTaskPlaceOnEventList(
76 &( pxQueue->xTasksWaitingToSend ), xTicksToWait );(19)
77 /* 佇列解鎖 */
78 prvUnlockQueue( pxQueue ); (20)
79
80 /* 恢復排程器 */
81 if ( xTaskResumeAll() == pdFALSE ) {
82 portYIELD_WITHIN_API();
83 }
84 } else {
85 /* 佇列有空閒訊息空間，允許入隊 */
86 prvUnlockQueue( pxQueue ); (21)
87 ( void ) xTaskResumeAll();
88 }
89 } else {
90 /* 超時時間已過，退出 */
91 prvUnlockQueue( pxQueue ); (22)
92 ( void ) xTaskResumeAll();
93
94 traceQUEUE_SEND_FAILED( pxQueue );
95 return errQUEUE_FULL;
96 }
97 }
98 }
99 /*-----------------------------------------------------------*/
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
如果阻塞時間不為 0，任務會因為等待入隊而進入阻塞， 在將任務設定為阻塞的過程中， 系統不希望有其它任務和中斷操作這個佇列的 xTasksWaitingToReceive 列表和 xTasksWaitingToSend 列表，因為可能引起其它任務解除阻塞，這可能會發生優先順序翻轉。比如任務 A 的優先順序低於當前任務，但是在當前任務進入阻塞的過程中，任務 A 卻因為其它原因解除阻塞了，這顯然是要絕對禁止的。因此FreeRTOS 使用掛起排程器禁止其它任務操作佇列，因為掛起排程器意味著任務不能切換並且不準呼叫可能引起任務切換的 API 函式。但掛起排程器並不會禁止中斷，中斷服務函式仍然可以操作佇列阻塞列表，可能會解除任務阻塞、可能會進行上下文切換，這也是不允許的。於是，FreeRTOS解決辦法是不但掛起排程器，還要給佇列上鎖，禁止任何中斷來操作佇列。
下面來看看流程圖：

相比在任務中呼叫的傳送函式，在中斷中呼叫的函式會更加簡單一些， 沒有任務阻塞操作。
函式 xQueueGenericSend中插入資料後， 會檢查等待接收連結串列是否有任務等待，如果有會恢復就緒。如果恢復的任務優先順序比當前任務高， 則會觸發任務切換；但是在中斷中呼叫的這個函式的做法是返回一個引數標誌是否需要觸發任務切換，並不在中斷中切換任務。
在任務中呼叫的函式中有鎖定和解鎖佇列的操作， 鎖定佇列的時候， 佇列的事件連結串列不能被修改。 而在被中斷中傳送訊息的處理是： 當遇到佇列被鎖定的時候， 將新資料插入到佇列後， 並不會直接恢復因為等待接收的任務， 而是累加了計數， 當佇列解鎖的時候， 會根據這個計數， 對應恢復幾個任務。
遇到佇列滿的情況， 函式會直接返回， 而不是阻塞等待， 因為在中斷中阻塞是不允許的！！！

1 BaseType_t xQueueGenericSendFromISR(
2 QueueHandle_t xQueue,
3 const void * const pvItemToQueue,
4 /* 不在中斷函式中觸發任務切換， 而是返回一個標記 */
5 BaseType_t * const pxHigherPriorityTaskWoken,
6 const BaseType_t xCopyPosition )
7{
8 BaseType_t xReturn;
9 UBaseType_t uxSavedInterruptStatus;
10 Queue_t * const pxQueue = ( Queue_t * ) xQueue;
11
12 uxSavedInterruptStatus = portSET_INTERRUPT_MASK_FROM_ISR();
13 {
14 // 判斷佇列是否有空間插入新內容
15 if( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength ) || ( xCopyPosition == queueOVERWRITE ) )
16 {
17 const int8_t cTxLock = pxQueue->cTxLock;
18
19 // 中斷中不能使用互斥鎖， 所以拷貝函式只是拷貝資料，
20 // 沒有任務優先順序繼承需要考慮
21 ( void ) prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition );
22
23 // 判斷佇列是否被鎖定
24 if( cTxLock == queueUNLOCKED )
25 {
26 #if ( configUSE_QUEUE_SETS == 1 )
27 // 集合相關程式碼
28 #else /* configUSE_QUEUE_SETS */
29 {
30 // 將最高優先順序的等待任務恢復到就緒連結串列
31 if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE )
32 {
33 if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE)
34 {
35 // 如果有高優先順序的任務被恢復
36 // 此處不直接觸發任務切換， 而是返回一個標記
37 if( pxHigherPriorityTaskWoken != NULL )
38 {
39 *pxHigherPriorityTaskWoken = pdTRUE;
40 }
41 }
42 }
43 }
44 #endif /* configUSE_QUEUE_SETS */
45 }
46 else
47 {
48 // 佇列被鎖定， 不能修改事件連結串列
49 // 增加計數， 記錄需要接觸幾個任務到就緒
50 // 在解鎖佇列的時候會根據這個計數恢復任務
51 pxQueue->cTxLock = ( int8_t ) ( cTxLock + 1 );
52 }
53 xReturn = pdPASS;
54 }
55 else
56 {
57 // 佇列滿 直接返回 不阻塞
58 xReturn = errQUEUE_FULL;
59 }
60 }
61
62 // 恢復中斷的優先順序
63 portCLEAR_INTERRUPT_MASK_FROM_ISR( uxSavedInterruptStatus );
64
65 return xReturn;
66}
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
訊息佇列讀取

任務呼叫接收函式收取佇列訊息， 函式首先判斷當前佇列是否有未讀訊息， 如果沒有， 則會判斷引數 xTicksToWait, 決定直接返回函式還是阻塞等待。
如果佇列中有訊息未讀， 首先會把待讀的訊息複製到傳進來的指標所指內， 然後判斷函式引數 xJustPeeking == pdFALSE的時候， 符合的話， 說明這個函式讀取了資料， 需要把被讀取的資料做出隊處理， 如果不是， 則只是檢視一下（peek），只是返回資料，但是不會把資料清除。
對於正常讀取資料的操作， 清除資料後佇列會空出空位， 所以檢視佇列中的等待列表中是否有任務等傳送資料而被掛起， 有的話恢復一個任務就緒， 並根據優先順序判斷是否需要出進行任務切換。
對於只是檢視資料的， 由於沒有清除資料， 所以沒有空間新空出，不需要檢查傳送等待連結串列， 但是會檢查接收等待連結串列， 如果有任務掛起會切換其到就緒並判斷是否需要切換。

訊息隊列出隊過程分析，其實跟入隊差不多，請看註釋：

1 /*-----------------------------------------------------------*/
2 BaseType_t xQueueGenericReceive( QueueHandle_t xQueue, (1)
3 void * const pvBuffer, (2)
4 TickType_t xTicksToWait, (3)
5 const BaseType_t xJustPeeking ) (4)
6 {
7 BaseType_t xEntryTimeSet = pdFALSE;
8 TimeOut_t xTimeOut;
9 int8_t *pcOriginalReadPosition;
10 Queue_t * const pxQueue = ( Queue_t * ) xQueue;
11
12 /* 已刪除一些斷言 */
13 for ( ;; ) {
14 taskENTER_CRITICAL(); (5)
15 {
16 const UBaseType_t uxMessagesWaiting = pxQueue->uxMessagesWaiting;
17
18 /* 看看佇列中有沒有訊息 */
19 if ( uxMessagesWaiting > ( UBaseType_t ) 0 ) { (6)
20 /*防止僅僅是讀取訊息，而不進行訊息出隊操作*/
21 pcOriginalReadPosition = pxQueue->u.pcReadFrom; (7)
22 /* 拷貝訊息到使用者指定存放區域pvBuffer */
23 prvCopyDataFromQueue( pxQueue, pvBuffer ); (8)
24
25 if ( xJustPeeking == pdFALSE ) { (9)
26 /* 讀取訊息並且訊息出隊 */
27 traceQUEUE_RECEIVE( pxQueue );
28
29 /* 獲取了訊息，當前訊息佇列的訊息個數需要減一 */
30 pxQueue->uxMessagesWaiting = uxMessagesWaiting - 1; (10)
31 /* 判斷一下訊息佇列中是否有等待發送訊息的任務 */
32 if ( listLIST_IS_EMPTY( (11)
33 &( pxQueue->xTasksWaitingToSend ) ) == pdFALSE ) {
34 /* 將任務從阻塞中恢復 */
35 if ( xTaskRemoveFromEventList( (12)
36 &( pxQueue->xTasksWaitingToSend ) ) != pdFALSE ) {
37 /* 如果被恢復的任務優先順序比當前任務高，會進行一次任務切換 */
38 queueYIELD_IF_USING_PREEMPTION(); (13)
39 } else {
40 mtCOVERAGE_TEST_MARKER();
41 }
42 } else {
43 mtCOVERAGE_TEST_MARKER();
44 }
45 } else { (14)
46 /* 任務只是看一下訊息（peek），並不出隊 */
47 traceQUEUE_PEEK( pxQueue );
48
49 /* 因為是隻讀訊息 所以還要還原讀訊息位置指標 */
50 pxQueue->u.pcReadFrom = pcOriginalReadPosition; (15)
51
52 /* 判斷一下訊息佇列中是否還有等待獲取訊息的任務 */
53 if ( listLIST_IS_EMPTY( (16)
54 &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) {
55 /* 將任務從阻塞中恢復 */
56 if ( xTaskRemoveFromEventList(
57 &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) {
58 /* 如果被恢復的任務優先順序比當前任務高，會進行一次任務切換 */
59 queueYIELD_IF_USING_PREEMPTION();
60 } else {
61 mtCOVERAGE_TEST_MARKER();
62 }
63 } else {
64 mtCOVERAGE_TEST_MARKER();
65 }
66 }
67
68 taskEXIT_CRITICAL(); (17)
69 return pdPASS;
70 } else { (18)
71 /* 訊息佇列中沒有訊息可讀 */
72 if ( xTicksToWait == ( TickType_t ) 0 ) { (19)
73 /* 不等待，直接返回 */
74 taskEXIT_CRITICAL();
75 traceQUEUE_RECEIVE_FAILED( pxQueue );
76 return errQUEUE_EMPTY;
77 } else if ( xEntryTimeSet == pdFALSE ) {
78 /* 初始化阻塞超時結構體變數，初始化進入
79 阻塞的時間xTickCount和溢位次數xNumOfOverflows */
80 vTaskSetTimeOutState( &xTimeOut ); (20)
81 xEntryTimeSet = pdTRUE;
82 } else {
83 mtCOVERAGE_TEST_MARKER();
84 }
85 }
86 }
87 taskEXIT_CRITICAL();
88
89 vTaskSuspendAll();
90 prvLockQueue( pxQueue ); (21)
91
92 /* 檢查超時時間是否已經過去了*/
93 if ( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) {(22)
94 /* 如果佇列還是空的 */
95 if ( prvIsQueueEmpty( pxQueue ) != pdFALSE ) {
96 traceBLOCKING_ON_QUEUE_RECEIVE( pxQueue ); (23)
97 /* 將當前任務新增到佇列的等待接收列表中
98 以及阻塞延時列表，阻塞時間為使用者指定的超時時間xTicksToWait */
99 vTaskPlaceOnEventList(
100 &( pxQueue->xTasksWaitingToReceive ), xTicksToWait );
101 prvUnlockQueue( pxQueue );
102 if ( xTaskResumeAll() == pdFALSE ) {
103 /* 如果有任務優先順序比當前任務高，會進行一次任務切換 */
104 portYIELD_WITHIN_API();
105 } else {
106 mtCOVERAGE_TEST_MARKER();
107 }
108 } else {
109 /* 如果佇列有訊息了，就再試一次獲取訊息 */
110 prvUnlockQueue( pxQueue ); (24)
111 ( void ) xTaskResumeAll();
112 }
113 } else {
114 /* 超時時間已過，退出 */
115 prvUnlockQueue( pxQueue ); (25)
116 ( void ) xTaskResumeAll();
117
118 if ( prvIsQueueEmpty( pxQueue ) != pdFALSE ) {
119 /* 如果佇列還是空的，返回錯誤程式碼errQUEUE_EMPTY */
120 traceQUEUE_RECEIVE_FAILED( pxQueue );
121 return errQUEUE_EMPTY; (26)
122 } else {
123 mtCOVERAGE_TEST_MARKER();
124 }
125 }
126 }
127 }
128 /*-----------------------------------------------------------*/
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
提示
如果佇列儲存的資料較大時，那最好是利用佇列來傳遞資料的指標而不是資料本身，因為傳遞資料的時候是需要CPU一位元組一位元組地將資料拷貝進佇列或從佇列拷貝出來。而傳遞指標無論是在處理速度上還是記憶體空間利用上都更有效。但是，當利用佇列傳遞指標時，一定要十分小心地做到以下兩點：

1.指標指向的記憶體空間的所有權必須明確
當任務間通過指標共享記憶體時，應該從根本上保證所不會有任意兩個任務同時修改共享記憶體中的資料，或是以其它行為方式使得共享記憶體資料無效或產生一致性問題。原則上，共享記憶體在其指標傳送到佇列之前，其內容只允許被髮送任務訪問；共享記憶體指標從佇列中被讀出之後，其內容亦只允許被接收任務訪問。

2.指標指向的記憶體空間必須有效
如果指標指向的記憶體空間是動態分配的，只應該有一個任務負責對其進行記憶體釋放。當這段記憶體空間被釋放之後，就不應該有任何一個任務再訪問這段空間。
並且最最最重要的是禁止使用指標訪問任務棧上的空間，也就是區域性變數。因為當棧發生改變後，棧上的資料將不再有效。