一直想體驗(yàn)下動(dòng)態(tài)消息隊(duì)列的特性，我一看見“ACE目前提供兩種動(dòng)態(tài)消息隊(duì)列：基于最終期限（deadline）的和基于松弛度（laxity）的（參見[IX]）動(dòng)態(tài)消息隊(duì)列”這句話，無限的好奇立即被激起，整了好久，結(jié)果我郁悶地發(fā)現(xiàn)不管我怎么設(shè)置，一點(diǎn)動(dòng)態(tài)的特性都沒有看見，和靜態(tài)消息隊(duì)列沒有任何不同。調(diào)試跟進(jìn)去才發(fā)現(xiàn)，原來在Windows和Linux的config文件中都沒有定義"ACE_HAS_TIMED_MESSAGE_BLOCKS"這個(gè)宏，所以msg_deadline_time和msg_execution_time都不起任何作用。懶得去重編ACE庫了，就學(xué)習(xí)靜態(tài)隊(duì)列的使用吧 ^_^
ACE_Message_Queue_Factory這個(gè)工廠提供三個(gè)靜態(tài)函數(shù)分別用來創(chuàng)建靜態(tài)消息隊(duì)列和兩種類型的動(dòng)態(tài)消息隊(duì)列。靜態(tài)消息隊(duì)列的消息也支持優(yōu)先級(jí)，但是消息的優(yōu)先級(jí)是靜態(tài)的，不需要通過動(dòng)態(tài)計(jì)算而來。消息隊(duì)列有兩個(gè)術(shù)語讓我困惑很久。水位用來控制消息隊(duì)列中數(shù)據(jù)的大小，高水位（high_water_mark）用于控制消息隊(duì)列的上限，它用于控制生產(chǎn)者往里面放數(shù)據(jù)的量，如果消息隊(duì)列中數(shù)據(jù)量已經(jīng)達(dá)到高水位，而用使用了鎖，既使用“ACE_Message_Queue_Factory<ACE_MT_SYNCH>::create_static_message_queue();”創(chuàng)建消息隊(duì)列，那么生產(chǎn)者將被阻塞。高水位很容易理解，但是低水位是用來做什么的呢？我本來地以為所謂的低水位嘛，肯定是用來控制消費(fèi)者的，比如低水位為20，那么當(dāng)數(shù)據(jù)量少于20時(shí)，消費(fèi)者將被阻塞，直到生產(chǎn)者往隊(duì)列中放入數(shù)據(jù)，數(shù)據(jù)量大于20時(shí)喚醒消費(fèi)者即可，消費(fèi)者此時(shí)可以將低水位調(diào)低比如到0，一次將這20個(gè)單位的數(shù)據(jù)消費(fèi)掉，然后再次將低水位調(diào)至20，下一次循環(huán)將自己阻塞，如此這般就可以很好地避免“傻窗口綜合癥”( silly window syndrome )。帶著這個(gè)理解去測(cè)試靜態(tài)隊(duì)列時(shí)讓我郁悶了好一陣子，看了一陣子源代碼之后才發(fā)現(xiàn)事實(shí)并不是這么回事。只要消息隊(duì)列中還有數(shù)據(jù)消費(fèi)者就不會(huì)被阻塞的，而當(dāng)數(shù)據(jù)量超過高水位時(shí)，生產(chǎn)者會(huì)被阻塞，既然會(huì)被阻塞，那么它肯定需要被喚醒，那么什么時(shí)候由誰來喚醒生產(chǎn)者呢？這就是低水位的作用，消費(fèi)者一直消費(fèi)數(shù)據(jù)，當(dāng)數(shù)據(jù)低于低水位時(shí)它就喚醒生產(chǎn)者。我感覺這似乎并不能解決傻窗口綜合癥，可能我是誤解了傻窗口綜合癥吧。
下面的代碼很好的展示了靜態(tài)消息隊(duì)列的使用。
#include "ace/Message_Queue.h"
#include "ace/Get_Opt.h"
#include "ace/OS.h"
#include <ace/Thread_Manager.h>
#include <ace/Synch.h>
//消息隊(duì)列指針
ACE_Message_Queue<ACE_MT_SYNCH>* mq;
const char S1[] = "C++";
const char S2[] = "Java";
const char S3[] = "PHP";
const char S4[] = "C#";
//四個(gè)消息指針
ACE_Message_Block* mb1, * mb2, * mb3, * mb4;
//生產(chǎn)者
static void* produce(void *arg)
{
static int loop = 1;
while(true)
{
ACE_OS::sleep(2);
ACE_DEBUG((LM_DEBUG, "(%P : %t) producer...\n"));
while(true)
{
if(loop == 1)
{
//將高水位設(shè)置為10， S1+S2的長度為3+4+2=9<10，因此可以將S3放進(jìn)去
//但是再放入S4時(shí)生產(chǎn)者將會(huì)被阻塞
//需要注意的是水位的大小并不是消息的個(gè)數(shù)，而是消息隊(duì)列中消息里面的數(shù)據(jù)量之和
//如果也能以消息的個(gè)數(shù)作為高低水位的值就好了
mq->high_water_mark(10);
mq->enqueue_prio (mb1);
mq->enqueue_prio (mb2);
mq->enqueue_prio (mb3);
ACE_DEBUG((LM_DEBUG, "(%P : %t) producer will pending!!\n"));
//因?yàn)橄M(fèi)者在睡眠6秒之后才會(huì)調(diào)用deactivate，因此生產(chǎn)者會(huì)在這兒阻塞幾秒鐘
//可以不斷地將msg_bytes打印出來觀察觀察
int ret = mq->enqueue_prio (mb4);
ACE_DEBUG((LM_DEBUG, "(%P : %t) producer waken up by deactivate, ret = %d!!\n", ret));
++loop;
}
if(loop == 2)
{
ACE_OS::sleep(6);
//將低水位設(shè)置為5，因?yàn)楦咚蝗匀粸?0，當(dāng)前的數(shù)據(jù)量又超過了10，
//所以下面的入隊(duì)操作仍會(huì)將生產(chǎn)者阻塞
//這樣消費(fèi)者消費(fèi)消息，當(dāng)數(shù)據(jù)量小于5時(shí)，將喚醒生產(chǎn)者
//生產(chǎn)者在此處等待被消費(fèi)者喚醒
mq->low_water_mark(5);
ACE_DEBUG((LM_DEBUG, "(%P : %t) producer will pending again!!\n"));
mq->enqueue_prio (mb4);
ACE_DEBUG((LM_DEBUG, "(%P : %t) producer waken up by consumer!!\n"));
++loop;
}
}
}
return NULL;
}
//消費(fèi)者
void* consume(void *arg)
{
static int loop = 1;
while(true)
{
ACE_OS::sleep(2);
ACE_DEBUG((LM_DEBUG, "(%P : %t) consumer...\n"));
if(loop == 1)
{
//等待6秒，此時(shí)生產(chǎn)者和消費(fèi)者都將被阻塞
ACE_OS::sleep(6);
//deactivate會(huì)喚醒所有的線程，將消息隊(duì)列設(shè)置為不可用
//以后所存取操作都會(huì)返回-1
//這個(gè)操作會(huì)喚醒生產(chǎn)者
mq->deactivate();
++loop;
}
if(loop == 2)
{
ACE_OS::sleep(2);
//將消息隊(duì)列的狀態(tài)設(shè)置成ACTIVATED
//消息又可以使用了
mq->activate();
++loop;
}
if(loop == 3)
{
ACE_OS::sleep(10);
//消費(fèi)兩個(gè)消息之后，數(shù)據(jù)量就小于5了，低于低水位將喚醒生產(chǎn)者
ACE_Message_Block *mb;
mq->dequeue_head (mb);
mq->dequeue_head (mb);
ACE_DEBUG((LM_DEBUG, "(%P : %t) consumer wake up producer!!\n"));
++loop;
}
}
return NULL;
}
int main(int argc, char* argv[])
{
mq = ACE_Message_Queue_Factory<ACE_MT_SYNCH>::create_static_message_queue();
int priority;
//使用隨機(jī)數(shù)作為消息的優(yōu)先級(jí)
//數(shù)字越高，優(yōu)先級(jí)越高
priority = ACE_OS::rand() % 100;
mb1 = new ACE_Message_Block(S1, sizeof S1, priority);
priority = ACE_OS::rand() % 100;
mb2 = new ACE_Message_Block(S2, sizeof S2, priority);
priority = ACE_OS::rand() % 100;
mb3 = new ACE_Message_Block(S3, sizeof S3, priority);
priority = ACE_OS::rand() % 100;
mb4 = new ACE_Message_Block(S4, sizeof S4, priority);
//將消息壓入隊(duì)列中，enqueue_prio根據(jù)消息的優(yōu)先級(jí)將消息放到適當(dāng)?shù)奈恢蒙?br /> //enqueue_head只是簡單地將數(shù)據(jù)存入隊(duì)列中，而不考慮消息的優(yōu)先級(jí)
//使用enqueue_prio壓入消息后，可以簡單通過dequeue_head和dequeue_tail
//分別按優(yōu)先級(jí)從高到低和從低到高取消息
//如果使用enqueue_head和enqueue_tail壓入消息
//則需要通過dequeue_prio來按照消息的優(yōu)先級(jí)依次將消息出隊(duì)列
//沒有必要既使用enqueue_prio壓入消息，又實(shí)用dequeue_prio來取消息
mq->enqueue_prio (mb1);
mq->enqueue_prio (mb2);
mq->enqueue_prio (mb3);
mq->enqueue_prio (mb4);
//輸出靜態(tài)消息隊(duì)列的相關(guān)信息
//高低水位默認(rèn)值均為16384
ACE_DEBUG((LM_DEBUG, "count : %d, bytes : %d, length : %d, high_water_mark : %d, low_water_mark : %d, status : %d\n",
mq->message_count(), mq->message_bytes(), mq->message_length(),
mq->high_water_mark(), mq->low_water_mark(),
mq->state()));
ACE_Message_Block *mb;
//使用next遍歷消息，遍歷的順序?yàn)楦邇?yōu)先級(jí)到底優(yōu)先級(jí)
ACE_DEBUG((LM_DEBUG, "===========next=============\n"));
//peek一下，并不彈出消息，類似Windows的PeekMessage
mq->peek_dequeue_head(mb);
do
{
ACE_DEBUG((LM_DEBUG, "message: %s, priority: %d\n", mb->rd_ptr(), mb->msg_priority()));
}while(mb = mb->next());
//使用迭代器遍歷消息隊(duì)列，遍歷的順序?yàn)楦邇?yōu)先級(jí)到底優(yōu)先級(jí)
ACE_DEBUG((LM_DEBUG, "=========iterator=============\n"));
ACE_Message_Queue<ACE_MT_SYNCH>::ITERATOR iterator (*mq);
for (ACE_Message_Block *entry = 0;
iterator.next (entry) != 0;
iterator.advance ())
{
ACE_DEBUG((LM_DEBUG, "message: %s, priority: %d\n", entry->rd_ptr(), entry->msg_priority()));
}
ACE_DEBUG((LM_DEBUG, "============dequeue_head==========\n"));
while(mq->dequeue_head (mb) != -1)
{
ACE_DEBUG((LM_DEBUG, "message: %s, priority: %d\n", mb->rd_ptr(), mb->msg_priority()));
//這里如果不判斷的話，消息隊(duì)列空時(shí)會(huì)導(dǎo)致主線程被阻塞
if(mq->is_empty())
break;
}
ACE_DEBUG((LM_DEBUG, "\n\n"));
//////////////////////////////////測(cè)試高低水位和隊(duì)列的state使用，進(jìn)行測(cè)試之前mq隊(duì)列已空///////////////////////////
//產(chǎn)生一個(gè)生產(chǎn)者線程
ACE_Thread_Manager::instance()->spawn_n
(
1,
(ACE_THR_FUNC) produce
);
////產(chǎn)生兩個(gè)消費(fèi)者線程
ACE_Thread_Manager::instance()->spawn_n
(
1,
(ACE_THR_FUNC) consume
);
//掛起主線程
ACE_Thread_Manager::instance()->wait();
return 0;
}
轉(zhuǎn)自
http://iknow.4ucode.com/Study/Topic/895400