簡介：

本文簡要介紹如何使用RT-Thread 的消息隊列 發(fā)送不定長數(shù)據(jù)的使用。數(shù)據(jù)比較經(jīng)典的用法，可以在很多數(shù)據(jù)傳輸?shù)膱龊蠎?yīng)用。

知識點：

1.參考內(nèi)容：

https://www.rt-thread.org/document/site/programming-manual/ipc2/ipc2/#_2

2.要點摘錄：通過定義結(jié)構(gòu)體：數(shù)據(jù)塊首地址和數(shù)據(jù)塊大小 來進行數(shù)據(jù)的傳遞。非常通用。

消息隊列的使用場合

消息隊列可以應(yīng)用于發(fā)送不定長消息的場合，包括線程與線程間的消息交換，以及中斷服務(wù)例程中給線程發(fā)送消息（中斷服務(wù)例程不能接收消息）。下面分發(fā)送消息和同步消息兩部分來介紹消息隊列的使用。

發(fā)送消息

消息隊列和郵箱的明顯不同是消息的長度并不限定在 4 個字節(jié)以內(nèi)；另外，消息隊列也包括了一個發(fā)送緊急消息的函數(shù)接口。但是當創(chuàng)建的是一個所有消息的最大長度是 4 字節(jié)的消息隊列時，消息隊列對象將蛻化成郵箱。這個不限定長度的消息，也及時的反應(yīng)到了代碼編寫的場合上，同樣是類似郵箱的代碼：

struct msg{
    rt_uint8_t *data_ptr;    /* 數(shù)據(jù)塊首地址 */
    rt_uint32_t data_size;   /* 數(shù)據(jù)塊大小   */};

和郵箱例子相同的消息結(jié)構(gòu)定義，假設(shè)依然需要發(fā)送這樣一個消息給接收線程。在郵箱例子中，這個結(jié)構(gòu)只能夠發(fā)送指向這個結(jié)構(gòu)的指針（在函數(shù)指針被發(fā)送過去后，接收線程能夠正確的訪問指向這個地址的內(nèi)容，通常這塊數(shù)據(jù)需要留給接收線程來釋放）。而使用消息隊列的方式則大不相同：

void send_op(void *data, rt_size_t length){    struct msg msg_ptr;

    msg_ptr.data_ptr = data;  /* 指向相應(yīng)的數(shù)據(jù)塊地址 */
    msg_ptr.data_size = length; /* 數(shù)據(jù)塊的長度 */

    /* 發(fā)送這個消息指針給 mq 消息隊列 */
    rt_mq_send(mq, (void*)&msg_ptr, sizeof(struct msg));
}

注意，上面的代碼中，是把一個局部變量的數(shù)據(jù)內(nèi)容發(fā)送到了消息隊列中。在接收線程中，同樣也采用局部變量進行消息接收的結(jié)構(gòu)體：

void message_handler(){    struct msg msg_ptr; /* 用于放置消息的局部變量 */

    /* 從消息隊列中接收消息到 msg_ptr 中 */
    if (rt_mq_recv(mq, (void*)&msg_ptr, sizeof(struct msg)) == RT_EOK)
    {        /* 成功接收到消息，進行相應(yīng)的數(shù)據(jù)處理 */
    }
}

因為消息隊列是直接的數(shù)據(jù)內(nèi)容復(fù)制，所以在上面的例子中，都采用了局部變量的方式保存消息結(jié)構(gòu)體，這樣也就免去動態(tài)內(nèi)存分配的煩惱了（也就不用擔心，接收線程在接收到消息時，消息內(nèi)存空間已經(jīng)被釋放）。

例子說明：

下面例子是通過onenet 平臺發(fā)送不定長命令，在開發(fā)板的命令響應(yīng)回掉函數(shù)中 將數(shù)據(jù)發(fā)送到打印線程：

/* onenet mqtt command response callback function */
static void onenet_cmd_rsp_cb(uint8_t *recv_data, size_t recv_size, uint8_t **resp_data, size_t *resp_size)
{
    char res_buf[] = { "cmd is received!\n" };

    log_d("recv data is %.*s\n", recv_size, recv_data);
    rs485_transport(recv_data,recv_size);
    /* user have to malloc memory for response data */
    *resp_data = (uint8_t *) ONENET_MALLOC(strlen(res_buf));

    strncpy(*resp_data, res_buf, strlen(res_buf));

    *resp_size = strlen(res_buf);
        
}
static rt_uint8_t temp_data[20];
void rs485_transport(uint8_t *recv_data, size_t recv_size)
{
          int result;
                struct msg msg_ptr;
             //增加一句 先  把數(shù)據(jù)復(fù)制到靜態(tài)變量中。
               strncpy(temp_data, recv_data, recv_size);
                msg_ptr.data_ptr = temp_data;//recv_data;
                msg_ptr.data_size = recv_size;
    rt_pin_write(RS485_RE_DE_Pin, PIN_HIGH);
/* 發(fā)送消息到消息隊列中 */
            result = rt_mq_send(&mq,(void*)&msg_ptr,sizeof(struct msg));
            if (result != RT_EOK)
            {
                rt_kprintf("rt_mq_send ERR\n");
            }

            rt_kprintf("thread onenet: send message - %s\n", msg_ptr.data_ptr);
    //rt_pin_write(RS485_RE_DE_Pin, PIN_LOW);
}

接收處理：
static void serial_thread_entry(void *parameter)
{
    char buf;
    rt_uint8_t k;
        struct msg msg_prt;
    while (1)
    {

                      /* 從消息隊列中接收消息 */

                if (rt_mq_recv(&mq, (void*)&msg_prt, sizeof(struct msg), RT_WAITING_FOREVER) == RT_EOK)
        {
        
                        rt_device_write(serial, 0, (void*)msg_prt.data_ptr, msg_prt.data_size);
        }
    
    }
}