學(xué)習(xí)RT-Thread的消息隊(duì)列，支持不定長(zhǎng)度消息的收發(fā)

前言

本文學(xué)習(xí)RT-Thread的消息隊(duì)列，支持不定長(zhǎng)度消息的收發(fā)，涉及消息隊(duì)列的工作機(jī)制、消息隊(duì)列相關(guān)函數(shù)以及基于STM32的RT-Thread消息隊(duì)列應(yīng)用示例，采用RTT&正點(diǎn)原子聯(lián)合出品潘多拉開發(fā)板進(jìn)行實(shí)驗(yàn)，基于STM32L475VET6。

一、消息隊(duì)列的工作機(jī)制

消息隊(duì)列能夠接收來(lái)自線程或中斷服務(wù)例程中不固定長(zhǎng)度的消息，并把消息緩存在自己的內(nèi)存空間中。其他線程也能夠從消息隊(duì)列中讀取相應(yīng)的消息，而當(dāng)消息隊(duì)列是空的時(shí)候，可以掛起讀取線程。當(dāng)有新的消息到達(dá)時(shí)，掛起的線程將被喚醒以接收并處理消息。消息隊(duì)列是一種異步的通信方式。

如下圖所示，線程或中斷服務(wù)例程可以將一條或多條消息放入消息隊(duì)列中。同樣，一個(gè)或多個(gè)線程也可以從消息隊(duì)列中獲得消息。當(dāng)有多個(gè)消息發(fā)送到消息隊(duì)列時(shí)，通常將先進(jìn)入消息隊(duì)列的消息先傳給線程，也就是說(shuō)，線程先得到的是最先進(jìn)入消息隊(duì)列的消息，即先進(jìn)先出原則 (FIFO)。

消息隊(duì)列工作示意圖（來(lái)源RT-Thread編程指南）

二、消息隊(duì)列的相關(guān)函數(shù)

1、創(chuàng)建動(dòng)態(tài)消息隊(duì)列函數(shù)：創(chuàng)建動(dòng)態(tài)消息隊(duì)列時(shí)先從對(duì)象管理器中分配一個(gè)消息隊(duì)列對(duì)象，然后給消息隊(duì)列對(duì)象分配一塊內(nèi)存空間，組織成空閑消息鏈表，這塊內(nèi)存的大小 =[消息大小 + 消息頭（用于鏈表連接）的大小]X 消息隊(duì)列最大個(gè)數(shù)，接著再初始化消息隊(duì)列，此時(shí)消息隊(duì)列為空。創(chuàng)建動(dòng)態(tài)消息隊(duì)列的函數(shù)接口如下所示：

1rt_mq_trt_mq_create(constchar*name,2rt_size_tmsg_size,3rt_size_tmax_msgs,4rt_uint8_tflag);

（1）入口參數(shù)：

name：消息隊(duì)列的名稱。msg_size：消息隊(duì)列中一條消息的最大長(zhǎng)度，單位字節(jié)。max_msgs：消息隊(duì)列的最大個(gè)數(shù)。flag：消息隊(duì)列采用的等待方式，它可以取如下數(shù)值：RT_IPC_FLAG_FIFO或RT_IPC_FLAG_PRIO。

（2）返回值：

RT_EOK：發(fā)送成功 消息隊(duì)列對(duì)象的句柄成功。RT_NULL：失敗。

2、刪除動(dòng)態(tài)消息隊(duì)列函數(shù)：當(dāng)動(dòng)態(tài)消息隊(duì)列不再被使用時(shí)，應(yīng)該刪除它以釋放系統(tǒng)資源，一旦操作完成，動(dòng)態(tài)消息隊(duì)列將被永久性地刪除。刪除動(dòng)態(tài)消息隊(duì)列時(shí)，如果有線程被掛起在該消息隊(duì)列等待隊(duì)列上，則內(nèi)核先喚醒掛起在該消息等待隊(duì)列上的所有線程（線程返回值是 - RT_ERROR），然后再釋放消息隊(duì)列使用的內(nèi)存，最后刪除消息隊(duì)列對(duì)象。函數(shù)接口如下：

1rt_err_trt_mq_delete(rt_mq_tmq);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。

（2）返回值：

RT_EOK：成功。

3、創(chuàng)建靜態(tài)消息隊(duì)列函數(shù)：創(chuàng)建靜態(tài)消息隊(duì)列和《RT-Thread編程指南》所講的初始化靜態(tài)消息隊(duì)列是一樣的，靜態(tài)消息隊(duì)列對(duì)象的內(nèi)存是在系統(tǒng)編譯時(shí)由編譯器分配的，一般放于讀數(shù)據(jù)段或未初始化數(shù)據(jù)段中。消息隊(duì)列初始化后所有消息都掛在空閑消息鏈表上，消息隊(duì)列為空，函數(shù)接口如下：

1rt_err_trt_mq_init(rt_mq_tmq,2constchar*name,3void*msgpool,4rt_size_tmsg_size,5rt_size_tpool_size,6rt_uint8_tflag);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。name：消息隊(duì)列的名稱。msgpool：指向存放消息的緩沖區(qū)的指針。msg_size：消息隊(duì)列中一條消息的最大長(zhǎng)度，單位字節(jié)。pool_size：存放消息的緩沖區(qū)大小。flag：消息隊(duì)列采用的等待方式，它可以取如下數(shù)值：RT_IPC_FLAG_FIFO 或RT_IPC_FLAG_PRIO。

（2）返回值：

RT_EOK：成功。

4、刪除靜態(tài)消息隊(duì)列函數(shù)：刪除靜態(tài)消息隊(duì)列和《RT-Thread編程指南》所講的脫離消息隊(duì)列是一樣的，脫離消息隊(duì)列將使消息隊(duì)列對(duì)象被從內(nèi)核對(duì)象管理器中脫離。內(nèi)核先喚醒所有掛在該消息等待隊(duì)列對(duì)象上的線程（線程返回值是 RT_ERROR），然后將該消息隊(duì)列對(duì)象從內(nèi)核對(duì)象管理器中脫離。函數(shù)接口如下：

1rt_err_trt_mq_detach(rt_mq_tmq);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。

（2）返回值：

RT_EOK：成功。

5、發(fā)送消息函數(shù)：線程或者中斷服務(wù)程序都可以給消息隊(duì)列發(fā)送消息。當(dāng)發(fā)送消息時(shí)，消息隊(duì)列對(duì)象先從空閑消息鏈表上取下一個(gè)空閑消息塊，把線程或者中斷服務(wù)程序發(fā)送的消息內(nèi)容復(fù)制到消息塊上，然后把該消息塊掛到消息隊(duì)列的尾部。當(dāng)且僅當(dāng)空閑消息鏈表上有可用的空閑消息塊時(shí)，發(fā)送者才能成功發(fā)送消息；當(dāng)空閑消息鏈表上無(wú)可用消息塊，說(shuō)明消息隊(duì)列已滿，此時(shí)，發(fā)送消息的的線程或者中斷程序會(huì)收到一個(gè)錯(cuò)誤碼（-RT_EFULL）。在發(fā)送一個(gè)普通消息之后，空閑消息鏈表上的隊(duì)首消息被轉(zhuǎn)移到了消息隊(duì)列尾。函數(shù)接口如下：

1rt_err_trt_mq_send(rt_mq_tmq,void*buffer,rt_size_tsize);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。buffer：消息內(nèi)容。size：消息大小。

（2）返回值：

RT_EOK：成功。RT_EFULL：消息隊(duì)列已滿。RT_ERROR：失敗，表示發(fā)送的消息長(zhǎng)度大于消息隊(duì)列中消息的最大長(zhǎng)度。

6、發(fā)送緊急消息函數(shù)：發(fā)送緊急消息的過(guò)程與發(fā)送消息幾乎一樣，唯一的不同是，當(dāng)發(fā)送緊急消息時(shí)，從空閑消息鏈表上取下來(lái)的消息塊不是掛到消息隊(duì)列的隊(duì)尾，而是掛到隊(duì)首，這樣，接收者就能夠優(yōu)先接收到緊急消息，從而及時(shí)進(jìn)行消息處理。發(fā)送緊急消息的函數(shù)接口如下：

1rt_err_trt_mq_urgent(rt_mq_tmq,void*buffer,rt_size_tsize);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。buffer：消息內(nèi)容。size：消息大小。

（2）返回值：

RT_EOK：成功。RT_EFULL：消息隊(duì)列已滿。RT_ERROR：失敗。

7、接收消息函數(shù)：當(dāng)消息隊(duì)列中有消息時(shí)，接收者才能接收消息，否則接收者會(huì)根據(jù)超時(shí)時(shí)間設(shè)置，或掛起在消息隊(duì)列的等待線程隊(duì)列上，或直接返回。接收消息時(shí)，接收者需指定存儲(chǔ)消息的消息隊(duì)列對(duì)象句柄，并且指定一個(gè)內(nèi)存緩沖區(qū)，接收到的消息內(nèi)容將被復(fù)制到該緩沖區(qū)里。此外，還需指定未能及時(shí)取到消息時(shí)的超時(shí)時(shí)間。接收一個(gè)消息后消息隊(duì)列上的隊(duì)首消息被轉(zhuǎn)移到了空閑消息鏈表的尾部。函數(shù)接口如下：

1rt_err_trt_mq_recv(rt_mq_tmq,2void*buffer,3rt_size_tsize,4rt_int32_ttimeout);

（1）入口參數(shù)：

mq：消息隊(duì)列對(duì)象的句柄。buffer：消息內(nèi)容。size：消息大小。timeout：指定的超時(shí)時(shí)間。

（2）返回值：

RT_EOK：成功收到。RT_ETIMEOUT：超時(shí)。RT_ERROR：失敗，返回錯(cuò)誤。

三、基于STM32的消息隊(duì)列應(yīng)用示例

前將了很多消息隊(duì)列的理論知識(shí)，光說(shuō)不練都是假把式，那么接下來(lái)我們就進(jìn)行實(shí)際操作，基于潘多拉開發(fā)板進(jìn)行實(shí)驗(yàn)。創(chuàng)建一個(gè)消息隊(duì)列和兩個(gè)線程，其中一個(gè)線程用于發(fā)送消息和發(fā)送緊急消息，另外一個(gè)線程用于接收消息。通過(guò)按下按鍵KEY0發(fā)送消息msg1，按下按鍵KEY1發(fā)送緊急消息msg2，按下按鍵KEY2接收消息，接收到消息后打印出來(lái)。

1、實(shí)現(xiàn)代碼：

1/*線程句柄*/ 2staticrt_thread_tthread1=RT_NULL; 3staticrt_thread_tthread2=RT_NULL; 4 5/*消息隊(duì)列句柄*/ 6staticrt_mq_tmsgqueue1=RT_NULL; 7 8 9/************************************************************** 10函數(shù)名稱:thread1_recv_msgqueue 11函數(shù)功能:線程1入口函數(shù)，用于接收消息 12輸入?yún)?shù):parameter：入口參數(shù) 13返回值:無(wú) 14備注:無(wú) 15**************************************************************/ 16voidthread1_recv_msgqueue(void*parameter) 17{ 18u8key; 19charbuf[64]; 20 21while(1) 22{ 23key=key_scan(0); 24 25if(key==KEY2_PRES) 26{ 27if(rt_mq_recv(msgqueue1,buf,sizeof(buf),RT_WAITING_FOREVER)==RT_EOK) 28{ 29rt_kprintf("recvmsg:%s\r\n",buf); 30} 31else 32{ 33rt_kprintf("recvmsgfailed\r\n"); 34} 35} 36 37rt_thread_mdelay(1); 38} 39} 40 41/************************************************************** 42函數(shù)名稱:thread2_send_mailbox_msg 43函數(shù)功能:線程2入口函數(shù)，用于發(fā)送消息 44輸入?yún)?shù):parameter：入口參數(shù) 45返回值:無(wú) 46備注:無(wú) 47**************************************************************/ 48voidthread2_send_msgqueue(void*parameter) 49{ 50u8key; 51rt_err_tres; 52 53while(1) 54{ 55key=key_scan(0); 56 57if(key==KEY0_PRES) 58{ 59res=rt_mq_send(msgqueue1,"msg1",sizeof("msg1")); 60 61if(res==RT_EOK) 62{ 63rt_kprintf("msgqueuesendmsg1successful\r\n"); 64} 65else 66{ 67rt_kprintf("msgqueuesendmsg1failed\r\n"); 68} 69} 70elseif(key==KEY1_PRES) 71{ 72rt_mq_urgent(msgqueue1,"msg2",sizeof("msg2")); 73 74if(res==RT_EOK) 75{ 76rt_kprintf("msgqueuesendmsg2successful\r\n"); 77} 78else 79{ 80rt_kprintf("msgqueuesendmsg2failed\r\n"); 81} 82} 83 84rt_thread_mdelay(1); 85} 86} 87 88 89voidrtthread_msgqueue_test(void) 90{ 91msgqueue1=rt_mq_create("msgqueue1", 9264,/*消息最大長(zhǎng)度*/ 9310,/*消息隊(duì)列最大容量*/ 94RT_IPC_FLAG_FIFO);/*FIFO模式*/ 95 96if(msgqueue1!=RT_NULL) 97{ 98rt_kprintf("RT-Threadcreatemsgqueuesuccessful\r\n"); 99}100else101{102rt_kprintf("RT-Threadcreatemsgqueuefailed\r\n");103return;104}105106thread1=rt_thread_create("thread1",107thread1_recv_msgqueue,108NULL,109512,1103,11120);112113if(thread1!=RT_NULL)114{115rt_thread_startup(thread1);;116}117else118{119rt_kprintf("createthread1failed\r\n");120return;121}122123thread2=rt_thread_create("thread2",124thread2_send_msgqueue,125NULL,126512,1272,12820);129130if(thread2!=RT_NULL)131{132rt_thread_startup(thread2);;133}134else135{136rt_kprintf("createthread2failed\r\n");137return;138}139140}

2、觀察FinSH：

（1）開機(jī)，按下KEY2接收消息，沒(méi)有任何反應(yīng)，為什么呢？因?yàn)榇藭r(shí)消息隊(duì)列為空，還沒(méi)有任何消息，輸入list_msgqueue可查看當(dāng)前的消息隊(duì)列有msgqueue1，消息隊(duì)列里面有0條消息，以及掛起等待消息的線程為thread1：

（2）按下KEY0發(fā)送消息msg1，提示發(fā)送成功以及打印出接收到msg1的內(nèi)容（因?yàn)榍懊娌襟E（1）我們已經(jīng)先按下KEY2然后thread1掛起等待接收消息了，所有為直接打印出msg1）：

（3）接著連續(xù)按下10次KEY0發(fā)送消息msg1，當(dāng)按下第11次時(shí)，發(fā)送消息msg1失敗，因?yàn)槌跏蓟㈥?duì)列時(shí)最大只有支持10條消息：

（4）連續(xù)按下10次KEY2接收消息：

（5）先按下兩次KEY0發(fā)送兩次消息msg1，再按下一次KEY1發(fā)送一次緊急消息msg2：

（6）連續(xù)按下3次KEY2接收消息，先接收到的是一條msg2，再是兩條msg1（因?yàn)榍懊娌襟E2按下KEY1是發(fā)送緊急消息，所以先接收到的是msg2）：

四、消息隊(duì)列的使用場(chǎng)合

消息隊(duì)列可以應(yīng)用于發(fā)送不定長(zhǎng)消息的場(chǎng)合，包括線程與線程間的消息交換，以及中斷服務(wù)例程中給線程發(fā)送消息（中斷服務(wù)例程不能接收消息）。

1、發(fā)送消息

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

1structmsg2{3rt_uint8_t*data_ptr;/*數(shù)據(jù)塊首地址*/4rt_uint32_tdata_size;/*數(shù)據(jù)塊大小*/5};

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

1voidsend_op(void*data,rt_size_tlength)2{3structmsgmsg_ptr;4msg_ptr.data_ptr=data;/*指向相應(yīng)的數(shù)據(jù)塊地址*/5msg_ptr.data_size=length;/*數(shù)據(jù)塊的長(zhǎng)度*/6/*發(fā)送這個(gè)消息指針給mq消息隊(duì)列*/7rt_mq_send(mq,(void*)&msg_ptr,sizeof(structmsg));8}

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

1voidmessage_handler() 2{ 3structmsgmsg_ptr;/*用于放置消息的局部變量*/ 4 5/*從消息隊(duì)列中接收消息到msg_ptr中*/ 6if(rt_mq_recv(mq,(void*)&msg_ptr,sizeof(structmsg))==RT_EOK) 7{ 8/*成功接收到消息，進(jìn)行相應(yīng)的數(shù)據(jù)處理*/ 9}10}

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

2、同步消息

在一般的系統(tǒng)設(shè)計(jì)中會(huì)經(jīng)常遇到要發(fā)送同步消息的問(wèn)題，這個(gè)時(shí)候就可以根據(jù)當(dāng)時(shí)狀態(tài)的不同選擇相應(yīng)的實(shí)現(xiàn)：兩個(gè)線程間可以采用 [消息隊(duì)列 + 信號(hào)量或郵箱] 的形式實(shí)現(xiàn)。發(fā)送線程通過(guò)消息發(fā)送的形式發(fā)送相應(yīng)的消息給消息隊(duì)列，發(fā)送完畢后希望獲得接收線程的收到確認(rèn)，工作示意圖如下圖所示：

同步消息示意圖（來(lái)源RT-Thread編程指南）

根據(jù)消息確認(rèn)的不同，可以把消息結(jié)構(gòu)體定義成：

（1）消息采用了信號(hào)量來(lái)作為確認(rèn)標(biāo)志。信號(hào)量作為確認(rèn)標(biāo)志只能夠單一的通知發(fā)送線程，消息已經(jīng)確認(rèn)接收

1structmsg 2{ 3/*消息結(jié)構(gòu)其他成員*/ 4structrt_semaphoreack; 5};

（2）消息使用了郵箱來(lái)作為確認(rèn)標(biāo)志，郵箱作為確認(rèn)標(biāo)志，代表著接收線程能夠通知一些狀態(tài)值給發(fā)送線程：