訊息佇列的原理

簡介

訊息驅動機制是 gui 系統的基礎，訊息驅動的底層基礎設施之一是訊息佇列，它是整個 gui 系統運轉中樞，本文介紹了乙個基於環形佇列的訊息佇列實現方法，給出了它的資料結構、主要操作流程和核心**。

環形佇列

環行佇列是一種首尾相連的佇列資料結構，遵循先進先出原則，如下圖所示：

ring buffer 示意圖

在環形佇列中用一組連續位址的儲存單元依次存放從佇列頭到佇列尾的元素，通過兩個指標 read_pos 和 write_pos 分別指向讀取位置和寫入位置。

初始化佇列時，令 read_pos = write_pos = 0，每當寫入乙個新元素時，write_pos 增 1；每當讀取乙個元素時，read_pos 增 1 。若佇列已滿，不能往佇列寫入資料；若隊列為空，則不能讀取資料。判斷對列是否為滿的的方法是看(write_pos + 1)% queue_size == read_pos 是否成立，判斷佇列是否為空的方法是看 write_pos == read_pos 是否成立。

鑑於多個執行緒同時訪問環形佇列，需要考慮執行緒之間的互斥和同步問題，擬採用鎖控制多個執行緒互斥訪問環形佇列，使用訊號量控制線程之間的同步。

一段時間內只能有乙個執行緒獲得鎖，當它持有鎖時，其它執行緒要訪問環形佇列必須等待，直到前者釋放鎖。由此，鎖可以保證多個執行緒互斥的訪問環形佇列。

執行緒從佇列對資料前首先判斷訊號量是否大於 1 ，若是，則從佇列讀資料；否則，進入等待狀態，直到訊號量大於 1 為止；執行緒往佇列寫入乙個資料後，會將訊號量增 1 ，若有執行緒在等待，則會被喚醒。由此，訊號量實現了多執行緒同步訪問環形佇列。

流程圖下圖是環形緩衝區的初始化、讀資料、寫資料的主要流程。

ring buffer 流程圖

初始化時為環形佇列分配記憶體空間，並完成鎖和訊號量的初始化；

若往環形佇列寫資料，首先要獲得鎖，若鎖已被占用，則進入等待狀態，否則進一步去判斷環形佇列是否已滿。若滿了，則釋放鎖並返回；若佇列未滿，將資料寫入 write_pos 位置write_pos 增 1，釋放鎖並將訊號量增 1，表示已寫入乙個資料；

若從環形佇列讀資料，首先判斷訊號量是否大於 1 ，若不是，則等待，否則去獲取鎖，若鎖已被占用，則等待，否則從 read_pos 位置讀取資料，將read_pos 增 1 ，釋放鎖，讀取完畢。

資料結構

環形佇列的資料結構如下所示：

01 typedef _msg msg;

05 06 typedef _msgque msgqueue;

環形佇列包括如下資料：

lock：互斥鎖；

wait：訊號量

msg：指向資料區的指標；

size：環形佇列資料最大個數；

read_ops：讀取位置；

write_ops：寫入位置。

佇列初始化

初始化主要完成三個任務：

為環形佇列分配記憶體；

初始化互斥鎖，用 pthread_mutex_init 完成；

初始化訊號量，用 sem_init 完成。

01 /* create message queue /

02 _msg_queue = malloc (sizeof (msgqueue));

03 04 / init lock and sem /

05 pthread_mutex_init (&_msg_queue->lock, null);

06 sem_init (&_msg_queue->wait, 0, 0);

07 08 / allocate message memory /

09 _msg_queue -> msg = malloc (sizeof(msg) * nr_msg);

10 _msg_queue -> size = nr_msg;

寫操作如上面的流程圖介紹，寫操作主要包括如下幾步：

獲取鎖；

判斷佇列是否已滿；

若沒滿，將資料寫入 write_pos 處，將 write_pos 增 1，並判斷 write_pos 是否越界；

釋放鎖，並將訊號量增 1。

01 / lock the message queue /

02 pthread_mutex_lock (_msg_queue->lock);

03 04 / check if the queue is full. /

05 if ((_msg_queue->write_pos + 1)% _msg_queue->size == _msg_queue->read_pos)

10 11 / write a data to write_pos. /

12 _msg_queue -> msg [write_pos] = msg;

13 write_pos ++;

14 15 / check if write_pos if overflow. /

16 if (_msg_queue->write_pos >= _msg_queue->size)

17 _msg_queue->write_pos = 0;

18 19 / release lock /

20 pthread_mutex_unlock (_msg_queue->lock);

21 22 sem_post (_msg_queue->wait);

讀操作同理，讀操作分如下幾個步驟：

檢查訊號量；

獲取鎖；

判斷佇列是否為空；

若不為空，則讀取 read_ops 處的資料，將 read_ops 增 1，並判斷 read_pos 是否越界；

並釋放鎖。

01 sem_wait (_msg_queue->wait);

02 03 / lock the message queue /

04 pthread_mutex_lock (_msg_queue->lock);

05 06 / check if queue is empty /

07 if (_msg_queue->read_pos != _msg_queue->write_pos)

17 18 / release lock/

19 pthread_mutex_unlock (_msg_queue->lock);

問題本文採用的環形佇列是固定長度的，還可進一步改進，設計成可變長度的環形佇列；

本文的訊息佇列是「先進先出」原則，沒有考慮帶優先順序的訊息，但這種場合是存在的；

本文重點介紹了訊息佇列的原理和實現，對於乙個 gui 程式來講，還需要乙個訊息迴圈與訊息佇列一起工作，訊息迴圈將單獨總結。

訊息佇列的原理

Android訊息佇列原理

訊息佇列之RabbitMQ原理

訊息佇列的通訊原理及建立訊息佇列程式設計收發資料

訊息佇列的原理

Android訊息佇列原理

訊息佇列之RabbitMQ原理

訊息佇列的通訊原理及建立訊息佇列程式設計收發資料

相關推薦