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

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

??????? PRQueue是使用c++的兩個(gè)STL的deque還有pthread線程庫(kù)實(shí)現(xiàn)的，并且在例子中使用了兩個(gè)簡(jiǎn)單的類(lèi)-Mutex和Lock來(lái)展示這個(gè)邏輯。StringMsg類(lèi)表現(xiàn)一個(gè)樣本消息，QueueTest類(lèi)用來(lái)測(cè)試。

??????? 我選擇STL的deque是因?yàn)閐eque擁有很多必要的操作(包括operator[])來(lái)實(shí)現(xiàn)PRQueue。特別的，有一點(diǎn)很重要的是push_back和pop_front()操作對(duì)于deque的元素的指針或引用來(lái)說(shuō)都是有效地。

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

??????? 因?yàn)橄⒌暮诵奶幚磉^(guò)程是發(fā)生在多線程中，消息的就緒順序也許跟原始的輸入隊(duì)列不同。例如：如果一個(gè)線程從輸入隊(duì)列中拿走了一個(gè)報(bào)文消息后進(jìn)入休眠狀態(tài)，另外一個(gè)線程取出下一個(gè)報(bào)文消息，在第一個(gè)線程之前就運(yùn)行完成處理這個(gè)報(bào)文消息，并把這個(gè)報(bào)文放入輸出隊(duì)列中。因此，這輸出日志也許會(huì)無(wú)序了（我們假設(shè)消息被處理完之后才日志）。

?????? 使用PRQueue 則以上這種場(chǎng)景就能避免，它將會(huì)確保在輸出隊(duì)列中報(bào)文的順序和輸入隊(duì)列中保持一致，而不管線程處理報(bào)文的順序如何。PRQueue的基本邏輯比較簡(jiǎn)單，當(dāng)下一個(gè)報(bào)文從輸入隊(duì)列中取出的時(shí)候，仍然放入鎖中，下一個(gè)push_back的位置。然后釋放鎖，并且繼續(xù)處理。在這個(gè)消息報(bào)文完全處理完之后，先前請(qǐng)求的位置用來(lái)把這個(gè)消息放入輸出隊(duì)列。

???? PRQueue 使用兩個(gè)隊(duì)列構(gòu)造：'data’ 和'filled’。

???? 'filled’隊(duì)列的一個(gè)元素使用數(shù)據(jù)填充并且能夠從PRQueue彈出。一個(gè)封裝類(lèi)DataQueue是'data’和'filled’隊(duì)列的裝載器。 這種設(shè)計(jì)允許我們?cè)诖_切的實(shí)現(xiàn)過(guò)程中分離線程安全代碼，因此用戶不需要關(guān)心任何鎖/解鎖的邏輯。

?????下面讓我們更詳細(xì)的討論P(yáng)RQUEUE。

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

???? 為了使用多線程測(cè)試PRQUEUE，PROCESS_MSG執(zhí)行。它從輸入隊(duì)列中取出一個(gè)StringMsg，通過(guò)調(diào)用StringMsg::process()方法來(lái)處理這個(gè)消息并且push這個(gè)報(bào)文。

//------------ 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方法不僅等待輸入隊(duì)列中下一個(gè)報(bào)文的到來(lái)，也通過(guò)查看’filled’隊(duì)列中元素來(lái)檢查這個(gè)報(bào)文是否準(zhǔn)備彈出了。如果數(shù)據(jù)還沒(méi)有填充，pop將繼續(xù)阻塞等待

POP方法的處理邏輯：

1.????? 鎖定輸入隊(duì)列

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

3.????? 鎖住輸出隊(duì)列

4.????? 保留輸出隊(duì)列底部位置

5.????? 解鎖輸出隊(duì)列

6.????? 解鎖輸入隊(duì)列

代碼段如下：

// 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方法拷貝數(shù)據(jù)到輸出隊(duì)列的保留位置并且設(shè)置'filled’指示為真。它也通過(guò)發(fā)送一個(gè)通知信號(hào)來(lái)釋放掉等待一個(gè)條件變量的線程。

代碼段如下：

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

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

//------------ 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; }//

總結(jié)

?????? 在多線程應(yīng)用程序中，當(dāng)處理的消息流順序需要保證的時(shí)候，本文所說(shuō)的預(yù)留位置隊(duì)列將會(huì)是有用的。PRQueue將會(huì)確保輸出隊(duì)列中報(bào)文順序同輸入隊(duì)列保持一致，因?yàn)樵谳敵鲫?duì)列中下一個(gè)push_back的位置在輸入隊(duì)列取出報(bào)文的時(shí)候就同步的保留了。當(dāng)報(bào)文消息處理完成之后，所保留的位置隨后將會(huì)被數(shù)據(jù)填充。

注：完整代碼于此處下載：http://accu.org/content/journals/ol101/prqueue.zip（譯此文時(shí)，時(shí)間較倉(cāng)促，因此譯文很粗糙，待時(shí)間較寬松時(shí)再細(xì)細(xì)校驗(yàn)）。

?

作者：lgp88 發(fā)表于2012-3-12 13:57:53 原文鏈接 閱讀：88 評(píng)論：0 查看評(píng)論

?

轉(zhuǎn)載于:https://www.cnblogs.com/moonlove/archive/2012/03/12/2509149.html

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。