譯自：?http://accu.org/var/uploads/journals/overload101.pdf

??????? 在多線程應用程序中，要求消息輸入隊列和消息輸出隊列順序要求保持一致，而忽略多線程并發處理的順序，這種情況是比較難處理的。在本文中，作者設計了一種新型解決方案：PRQueue（預留位置隊列），較很好的解決這個問題。

??????? PRQueue是使用c++的兩個STL的deque還有pthread線程庫實現的，并且在例子中使用了兩個簡單的類-Mutex和Lock來展示這個邏輯。StringMsg類表現一個樣本消息，QueueTest類用來測試。

??????? 我選擇STL的deque是因為deque擁有很多必要的操作(包括operator[])來實現PRQueue。特別的，有一點很重要的是push_back和pop_front()操作對于deque的元素的指針或引用來說都是有效地。

??????? 這里有一個簡單的例子來展示PRQueue的作用。首先，我們需要記錄大量的多域的消息流。轉化文本字符串的數字域是一個慢的且并不關鍵的過程，因此我們決定將這份任務分派給能夠生成日志的線程來做。處理的流程如下圖所示：

??????? 因為消息的核心處理過程是發生在多線程中，消息的就緒順序也許跟原始的輸入隊列不同。例如：如果一個線程從輸入隊列中拿走了一個報文消息后進入休眠狀態，另外一個線程取出下一個報文消息，在第一個線程之前就運行完成處理這個報文消息，并把這個報文放入輸出隊列中。因此，這輸出日志也許會無序了（我們假設消息被處理完之后才日志）。

?????? 使用PRQueue 則以上這種場景就能避免，它將會確保在輸出隊列中報文的順序和輸入隊列中保持一致，而不管線程處理報文的順序如何。PRQueue的基本邏輯比較簡單，當下一個報文從輸入隊列中取出的時候，仍然放入鎖中，下一個push_back的位置。然后釋放鎖，并且繼續處理。在這個消息報文完全處理完之后，先前請求的位置用來把這個消息放入輸出隊列。

???? PRQueue 使用兩個隊列構造：'data’ 和'filled’。

???? 'filled’隊列的一個元素使用數據填充并且能夠從PRQueue彈出。一個封裝類DataQueue是'data’和'filled’隊列的裝載器。 這種設計允許我們在確切的實現過程中分離線程安全代碼，因此用戶不需要關心任何鎖/解鎖的邏輯。

?????下面讓我們更詳細的討論PRQUEUE。

?????PRQueue方法主要做兩件事情：它從輸入隊列中彈出數據，并且在輸出隊列中保留一個位置。PUSH方法使用之前保存的位置在輸出隊列中保存數據。

???? 為了使用多線程測試PRQUEUE，PROCESS_MSG執行。它從輸入隊列中取出一個StringMsg，通過調用StringMsg::process()方法來處理這個消息并且push這個報文。

//------------ static void* process_msg( void* arg) {int thidx = ++Thidx;QueueTest* quetest =(QueueTest*)arg;Msg* msg;PRQueue< Msg*>::position pos;cout << "Input thread=" << thidx << " started" << endl;for( ;;){// Wait for next message appeared in input queue,// pop it up and get push position allocated in output queuequetest->input_que.pop( msg, quetest->output_que, pos);// Process messagemsg->process( thidx);// Push processed message into output queue using acquired position quetest->output_que.push( msg, pos);} return NULL; }//

POP方法不僅等待輸入隊列中下一個報文的到來，也通過查看’filled’隊列中元素來檢查這個報文是否準備彈出了。如果數據還沒有填充，pop將繼續阻塞等待

POP方法的處理邏輯：

1.????? 鎖定輸入隊列

2.????? 如果輸入隊列非空并且頂層元素填充了數據，則pop它（否則釋放掉鎖并繼續睡眠）

3.????? 鎖住輸出隊列

4.????? 保留輸出隊列底部位置

5.????? 解鎖輸出隊列

6.????? 解鎖輸入隊列

代碼段如下：

// Pop data from input queue and reserve position in the output queuevoid pop( DT& data, PRQueue& outque, position& pos){Lock lk( m_mux);while( true) {if( m_que.pop( data))break;wait_while_empty();}outque.reserve_pos( pos);}

PUSH方法拷貝數據到輸出隊列的保留位置并且設置'filled’指示為真。它也通過發送一個通知信號來釋放掉等待一個條件變量的線程。

代碼段如下：

// Simple pushvoid push( const DT& data) {Lock lk( m_mux);m_que.push( data);notify_not_empty();}

?????? 現在，這個消息報文按序的到達了輸出隊列，如果我們想要更深的擴展我們的處理鏈的話，可以在后面再加上一個PRQueue。在以上的測試用例中我們不會這樣做：我們使用一個單一的線程簡單的從輸出隊列中讀取處理完的報文并將它們打印出來。在最后的一步，只簡單的使用了pop方法（未使用第二、第三個參數：指向輸出隊列和保留的位置的值）。

//------------ static void* print_msg( void* arg) {QueueTest* quetest =(QueueTest*)arg;Msg* msg;cout << "Output thread started" << endl;for( ;;){quetest->output_que.pop( msg);msg->print();delete msg;} return NULL; }//

總結

?????? 在多線程應用程序中，當處理的消息流順序需要保證的時候，本文所說的預留位置隊列將會是有用的。PRQueue將會確保輸出隊列中報文順序同輸入隊列保持一致，因為在輸出隊列中下一個push_back的位置在輸入隊列取出報文的時候就同步的保留了。當報文消息處理完成之后，所保留的位置隨后將會被數據填充。

注：完整代碼于此處下載：http://accu.org/content/journals/ol101/prqueue.zip（譯此文時，時間較倉促，因此譯文很粗糙，待時間較寬松時再細細校驗）。

?

作者：lgp88 發表于2012-3-12 13:57:53 原文鏈接 閱讀：88 評論：0 查看評論

?

轉載于:https://www.cnblogs.com/moonlove/archive/2012/03/12/2509149.html

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的[译]预留位置队列PRQueue：多线程程序中消息输入队列和消息输出队列保持同序...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。