作者：zzssdd2

E-mail：[email protected]

一、應用簡介

訊息佇列是RTOS中常用的一種資料通信方式，常用于任務與任務之間或是中斷與任務之間的資料傳遞，在裸機系統中我們通常會使用全域變數的方式進行資料傳遞，比如在事件發生后在中斷中改變資料和設定標志，然后在主回圈中輪詢不同的標志是否生效來對全域資料執行不同的操作，執行完畢后清除相關標志，但是這種方式需要不斷地輪詢標志狀態，使得CPU的利用率并不高，而使用RTOS的訊息佇列則具有任務阻塞機制，當沒有需要處理的訊息時任務掛起等待訊息，此時其他任務占用CPU執行其他操作，當有訊息放入佇列時任務恢復運行進行訊息接收和處理，這種訊息處理機制相比裸機而言大大地提高了CPU利用率，

ThreadX的訊息佇列支持“訊息置頂通知”功能，也就是可以將訊息放在佇列的最前面，使得任務可以及時處理某些緊急訊息（RT-Thread的訊息佇列也有該功能）
ThreadX的訊息佇列可以傳遞任意長度的資料，因為它是采用傳遞資料指標的方式（uCOS也是采用這種參考傳遞的方式，而FreeRTOS和RT-Thread則支持傳遞整體資料內容，這兩種方式各有優劣吧，指標傳遞方式優點是執行效率高，缺點是存資料的記憶體區域如果資料還未及時處理就被覆寫了那么就會引發問題；整體資料傳遞方式優點是安全不需擔心資料覆寫致錯，缺點是資料量大的話傳遞資料程序執行時間長導致效率低）

二、API簡介

下面介紹使用ThreadX的訊息佇列時常用的幾個API函式，

1、創建訊息佇列

描述

該服務用于創建訊息佇列，訊息總數是根據指定的訊息大小和佇列中的位元組總數來計算的

如果在佇列的記憶體區域中指定的位元組總數不能被指定的訊息大小均分，則不會使用該記憶體區域中的其余位元組

引數

queue_ptr 指向訊息佇列控制塊的指標

name_ptr 指向訊息佇列名稱的指標

message_size 指定佇列中每條訊息的大小，訊息大小選項為1個32位字到16個32位字之間（包含）

queue_start 訊息佇列的起始地址，起始地址必須與ULONG資料型別的大小對齊

queue_size 訊息佇列可用的位元組總數

回傳值

TX_SUCCESS (0x00) 創建成功

TX_QUEUE_ERROR (0x09) 無效的訊息佇列指標，指標為NULL或佇列已創建

TX_PTR_ERROR (0x03) 訊息佇列的起始地址無效

TX_SIZE_ERROR (0x05) 訊息佇列大小無效

TX_CALLER_ERROR (0x13) 該服務的呼叫者無效

UINT tx_queue_create(
    TX_QUEUE *queue_ptr, 
    CHAR *name_ptr,
    UINT message_size,
    VOID *queue_start, 
    ULONG queue_size);

2、洗掉訊息佇列

描述

此服務洗掉指定的訊息佇列，所有掛起等待此佇列訊息的執行緒都將恢復，并給出TX_DELETED回傳狀態

在洗掉佇列之前，應用程式必須確保已完成（或禁用）此佇列的所有send_notify回呼，此外，應用程式必須防止將來使用已洗掉的佇列

應用程式還負責管理與佇列相關聯的記憶體區域，該記憶體區域在此服務完成后可用

引數

queue_ptr 指向先前創建的訊息佇列的指標

回傳值

TX_SUCCESS (0x00) 洗掉成功

TX_QUEUE_ERROR (0x09) 訊息佇列指標無效

TX_CALLER_ERROR (0x13) 該服務的呼叫者無效

UINT tx_queue_delete(TX_QUEUE *queue_ptr);

3、清空訊息佇列

描述

此服務洗掉存盤在指定訊息佇列中的所有訊息

如果佇列已滿，將丟棄所有掛起執行緒的訊息，然后恢復每個掛起的執行緒，并回傳一個指示訊息發送成功的回傳狀態，如果佇列為空，則此服務不執行任何操作，

引數

queue_ptr 指向先前創建的訊息佇列的指標

回傳值

TX_SUCCESS (0x00) 操作成功

TX_QUEUE_ERROR (0x09) 訊息佇列指標無效

UINT tx_queue_flush(TX_QUEUE *queue_ptr);

4、訊息置頂

描述

該服務將訊息發送到指定訊息佇列的最前面，訊息從源指標指定的存盤區域復制到佇列的最前面

引數

queue_ptr 指向訊息佇列控制塊的指標

source_ptr 指向存放訊息的指標

wait_option 定義訊息佇列已滿時服務的行為

TX_NO_WAIT (0x00000000) - 無論是否成功都立即回傳（用于非執行緒呼叫，例如中斷里面）

TX_WAIT_FOREVER (0xFFFFFFFF) - 一直等待直到訊息佇列有空閑為止

回傳值

TX_SUCCESS (0x00) 操作成功

TX_DELETED (0x01) 執行緒掛起時，訊息佇列被洗掉

TX_QUEUE_FULL (0x0B) 服務無法發送訊息，因為在指定的等待時間內佇列已滿

TX_WAIT_ABORTED (0x1A) 被另一個執行緒、計時器或ISR中斷給中止

TX_QUEUE_ERROR (0x09) 無效的訊息佇列指標

TX_PTR_ERROR (0x03) 訊息的源指標無效

TX_WAIT_ERROR (0x04) 在非執行緒呼叫中指定了TX_NO_WAIT以外的等待選項

UINT tx_queue_front_send(
    TX_QUEUE *queue_ptr,
    VOID *source_ptr, 
    ULONG wait_option);

5、獲取訊息佇列資訊

描述

該服務檢索有關指定訊息佇列的資訊

引數（TX_NULL表示不需要獲取該引數代表的資訊）

queue_ptr 指向先前創建的訊息佇列的指標

name 指向目標的指標，用于指向佇列名稱

enqueued 指向目標的指標，表示當前佇列中的訊息數

available_storage 指向目標的指標，表示佇列當前有空間容納的訊息數

first_suspended 指向目標的指標，該指標指向該佇列的掛起串列中第一個執行緒

suspended_count 指向目標的指標，用于指示當前在此佇列上掛起的執行緒數

next_queue 指向下一個創建佇列的指標的目標的指標

回傳值

TX_SUCCESS (0x00) 操作成功

TX_QUEUE_ERROR (0x09) 無效的訊息佇列指標

UINT tx_queue_info_get(
    TX_QUEUE *queue_ptr, 
    CHAR **name,
    ULONG *enqueued, 
    ULONG *available_storage
    TX_THREAD **first_suspended, 
    ULONG *suspended_count,
    TX_QUEUE **next_queue);

6、從佇列獲取訊息

描述

該服務從指定的訊息佇列中檢索訊息，檢索到的訊息從佇列復制到目標指標指定的存盤區域，然后將該訊息從佇列中洗掉

指定的目標存盤區必須足夠大以容納訊息，也就是說，由destination_ptr 指向的訊息目標必須至少與此佇列的訊息大小一樣大，否則，如果目標不夠大，則會在存盤區域中發生記憶體地址非法錯誤

引數

queue_ptr 指向先前創建的訊息佇列的指標

destination_ptr 指向儲存訊息的地址

wait_option 定義訊息佇列為空時服務的行為

TX_NO_WAIT (0x00000000) - 無論是否成功都立即回傳（用于非執行緒呼叫，例如中斷里面）

TX_WAIT_FOREVER (0xFFFFFFFF) - 一直等待直到有訊息可以獲取

0x00000001 ~ 0xFFFFFFFE- 指定具體等待心跳節拍數（如果心跳頻率1KHZ，那么單位就是ms ）

回傳值

TX_SUCCESS (0x00) 操作成功

TX_DELETED (0x01) 執行緒掛起時洗掉了訊息佇列

TX_QUEUE_EMPTY (0x0A) 服務無法檢索訊息，因為佇列在指定的等待時間段內為空

TX_WAIT_ABORTED (0x1A) 被另一個執行緒、計時器或ISR中斷給中止

TX_QUEUE_ERROR (0x09) 無效的訊息佇列指標

TX_PTR_ERROR (0x03) 訊息的目標指標無效

TX_WAIT_ERROR (0x04) 在非執行緒呼叫中指定了TX_NO_WAIT以外的等待選項

UINT tx_queue_receive(
    TX_QUEUE *queue_ptr,
    VOID *destination_ptr, 
    ULONG wait_option);

7、向佇列發送訊息

描述

此服務將訊息發送到指定的訊息佇列，發送的訊息將從源指標指定的記憶體區域復制到佇列中，

引數

queue_ptr 指向先前創建的訊息佇列的指標

source_ptr 指向訊息的指標

wait_option 定義訊息佇列已滿時服務的行為

TX_NO_WAIT (0x00000000) - 無論是否成功都立即回傳（用于非執行緒呼叫，例如中斷里面）

TX_WAIT_FOREVER (0xFFFFFFFF) - 一直等待直到佇列有空位可以放置訊息

0x00000001 ~ 0xFFFFFFFE - 指定具體等待心跳節拍數（如果心跳頻率1KHZ，那么單位就是ms ）

回傳值

TX_SUCCESS (0x00) 操作成功

TX_DELETED (0x01) 執行緒掛起時洗掉了訊息佇列

TX_QUEUE_FULL (0x0B) 服務無法發送訊息，因為佇列在指定的等待時間內已滿

TX_WAIT_ABORTED (0x1A) 被另一個執行緒、計時器或ISR中斷給中止

TX_QUEUE_ERROR (0x09) 無效的訊息佇列指標

TX_PTR_ERROR (0x03) 訊息的目標指標無效

TX_WAIT_ERROR (0x04) 在非執行緒呼叫中指定了TX_NO_WAIT以外的等待選項

UINT tx_queue_send(
    TX_QUEUE *queue_ptr,
    VOID *source_ptr, 
    ULONG wait_option);

8、注冊發送通知回呼函式

描述

此服務注冊一個通知回呼函式，每當一條訊息發送到指定的佇列時就會呼叫該函式，通知回呼的處理由應用程式定義

不允許在應用程式的佇列發送通知回呼函式中呼叫具有暫停選項的ThreadX API

引數

queue_ptr 指向先前創建的佇列的指標

queue_send_notify 指向應用程式佇列發送通知功能的指標，如果此值為TX_NULL，則禁用通知

回傳值

TX_SUCCESS (0x00) 操作成功

TX_QUEUE_ERROR (0x09) 無效的佇列指標

TX_FEATURE_NOT_ENABLED (0xFF) 禁用了通知功能

UINT tx_queue_send_notify(
    TX_QUEUE *queue_ptr,
    VOID (*queue_send_notify)(TX_QUEUE *));

三、實體演示

該應用實體創建三個任務和一個佇列訊息發送通知回呼

任務1：按鍵1按一次向訊息佇列1發送一條訊息（單個變數訊息）

任務2：按鍵2按一次向訊息佇列2發送一條訊息（結構體指標訊息）

任務3：向訊息佇列3發送訊息；接收任務1和任務2的訊息并列印輸出訊息內容

回呼功能：輸出訊息佇列3的相關資訊

創建訊息佇列

#define DEMO_STACK_SIZE         (2 * 1024)
#define DEMO_BYTE_POOL_SIZE     (32 * 1024)

TX_THREAD       thread_0;
TX_THREAD       thread_1;
TX_THREAD       thread_2;

TX_BYTE_POOL    byte_pool_0;
UCHAR           memory_area[DEMO_BYTE_POOL_SIZE];

/* 訊息佇列 */
TX_QUEUE        tx_queue1;
TX_QUEUE        tx_queue2;
TX_QUEUE        tx_queue3;

ULONG           msg_queue1[32];
ULONG           msg_queue2[16];
ULONG           msg_queue3[8];

struct S_DATA{
    uint32_t id;
    uint16_t flag;
    uint8_t msg[2];
};
struct S_DATA data_package;

void thread_0_entry(ULONG thread_input);
void thread_1_entry(ULONG thread_input);
void thread_2_entry(ULONG thread_input);
void queue3_send_notify(TX_QUEUE *input);

void tx_application_define(void *first_unused_memory)
{
    CHAR    *pointer = TX_NULL;

    /* Create a byte memory pool from which to allocate the thread stacks. */
    tx_byte_pool_create(&byte_pool_0, "byte pool 0", memory_area, DEMO_BYTE_POOL_SIZE);

    /* Allocate the stack for thread 0. */
    tx_byte_allocate(&byte_pool_0, (VOID **) &pointer, DEMO_STACK_SIZE, TX_NO_WAIT);
    /* Create the main thread. */
    tx_thread_create(&thread_0, "thread 0", thread_0_entry, 0,  
                    pointer, DEMO_STACK_SIZE, 
                    1, 1, TX_NO_TIME_SLICE, TX_AUTO_START);

    /* Allocate the stack for thread 1. */
    tx_byte_allocate(&byte_pool_0, (VOID **) &pointer, DEMO_STACK_SIZE, TX_NO_WAIT);
    /* Create threads 1 */
    tx_thread_create(&thread_1, "thread 1", thread_1_entry, 0,  
                    pointer, DEMO_STACK_SIZE, 
                    2, 2, TX_NO_TIME_SLICE, TX_AUTO_START);

    /* Allocate the stack for thread 2. */
    tx_byte_allocate(&byte_pool_0, (VOID **) &pointer, DEMO_STACK_SIZE, TX_NO_WAIT);
    /* Create threads 1 */
    tx_thread_create(&thread_2, "thread 2", thread_2_entry, 0,  
                    pointer, DEMO_STACK_SIZE, 
                    3, 3, TX_NO_TIME_SLICE, TX_AUTO_START);

    /* 創建訊息佇列 */
    tx_queue_create(&tx_queue1, "tx_queue1", 1, msg_queue1, sizeof(msg_queue1));
    tx_queue_create(&tx_queue2, "tx_queue2", 1, msg_queue2, sizeof(msg_queue2));
    tx_queue_create(&tx_queue3, "tx_queue2", 1, msg_queue3, sizeof(msg_queue3));
    
    /* 注冊發送訊息回呼 */
    tx_queue_send_notify(&tx_queue3, queue3_send_notify);
}

任務1

void    thread_0_entry(ULONG thread_input)
{
    uint8_t i =0, key_flag = 0;
    uint8_t data_single = 0;

    while(1)
    {
        if (0 == key_flag)
        {
            if (GPIO_PIN_SET == HAL_GPIO_ReadPin(KEY1_GPIO_Port, KEY1_Pin))
            {
                key_flag = 1;
            }
        }
        else
        {
            if (GPIO_PIN_RESET == HAL_GPIO_ReadPin(KEY1_GPIO_Port,KEY1_Pin))
            {
                key_flag = 0;
                /*按鍵1觸發，向佇列1發送訊息*/
                data_single++;
                tx_queue_send(&tx_queue1, &data_single, TX_NO_WAIT);
            }
        }
        tx_thread_sleep(20);
    }
}

任務2

void    thread_1_entry(ULONG thread_input)
{
    uint8_t key_flag = 0;
    struct S_DATA *pData;

    pData = https://www.cnblogs.com/zzssdd2/archive/2020/12/22/&data_package;
    pData->id       = 1;
    pData->flag     = 2;
    pData->msg[0]   = 3;
    pData->msg[1]   = 4;

    while(1)
    {
        if (0 == key_flag)
        {
            if (GPIO_PIN_SET == HAL_GPIO_ReadPin(KEY2_GPIO_Port, KEY2_Pin))
            {
                key_flag = 1;
            }
        }
        else
        {
            if (GPIO_PIN_RESET == HAL_GPIO_ReadPin(KEY2_GPIO_Port,KEY2_Pin))
            {
                key_flag = 0;
                /*按鍵2觸發，向佇列2發送訊息*/
                pData->id       += 8;
                pData->flag     += 4;
                pData->msg[0]   += 2;
                pData->msg[1]   += 1;
                tx_queue_send(&tx_queue2, &pData, TX_NO_WAIT);
            }
        }
        tx_thread_sleep(20);
    }
}

任務3

void    thread_2_entry(ULONG thread_input)
{
    UINT status;
    uint8_t char_data;
    ULONG long_data = https://www.cnblogs.com/zzssdd2/archive/2020/12/22/0;
    struct S_DATA *buf_data;

    while(1)
    {
        /* 向佇列3發送訊息 */
        long_data++;
        tx_queue_send(&tx_queue3, &long_data, TX_NO_WAIT);
        if (0 == (long_data & 7))
        {
            tx_queue_flush(&tx_queue3);
        }

        /* 接收佇列1訊息 */
        status = tx_queue_receive(&tx_queue1, &char_data, 1000);
        if (TX_SUCCESS == status)
        {
            SEGGER_RTT_SetTerminal(0);
            SEGGER_RTT_printf(0, RTT_CTRL_TEXT_BRIGHT_GREEN"message queue1 receive data is %d\r\n", char_data);
        }

        /* 接收佇列2訊息 */
        status = tx_queue_receive(&tx_queue2, &buf_data, 1000);
        if (TX_SUCCESS == status)
        {
            SEGGER_RTT_SetTerminal(1);
            SEGGER_RTT_printf(0, RTT_CTRL_TEXT_BRIGHT_YELLOW"message queue2 receive data is %d\t%d\t%d\t%d \r\n", \
                                buf_data->id, \
                                buf_data->flag, \
                                buf_data->msg[0], \
                                buf_data->msg[1]);
        }
    } 	
}

發送佇列訊息回呼功能

void    queue3_send_notify(TX_QUEUE *input)
{
    ULONG enqueued;             // 佇列中的訊息數
    ULONG available_storage;    // 佇列剩余空間
    
    tx_queue_info_get(&tx_queue3, TX_NULL, &enqueued, &available_storage, TX_NULL, TX_NULL, TX_NULL);
    
    SEGGER_RTT_SetTerminal(2);
    SEGGER_RTT_printf(0, "the number of messages in the queue3 %d\r\n", enqueued);
    SEGGER_RTT_printf(0, "the queue3 remaining size %d\r\n", available_storage);
}

任務1演示結果