一直想體驗下動態消息隊列的特性，我一看見“ACE目前提供兩種動態消息隊列：基于最終期限（deadline）的和基于松弛度（laxity）的（參見[IX]）動態消息隊列”這句話，無限的好奇立即被激起，整了好久，結果我郁悶地發現不管我怎么設置，一點動態的特性都沒有看見，和靜態消息隊列沒有任何不同。調試跟進去才發現，原來在Windows和Linux的config文件中都沒有定義"ACE_HAS_TIMED_MESSAGE_BLOCKS"這個宏，所以msg_deadline_time和msg_execution_time都不起任何作用。懶得去重編ACE庫了，就學習靜態隊列的使用吧 ^_^

        ACE_Message_Queue_Factory這個工廠提供三個靜態函數分別用來創建靜態消息隊列和兩種類型的動態消息隊列。靜態消息隊列的消息也支持優先級，但是消息的優先級是靜態的，不需要通過動態計算而來。消息隊列有兩個術語讓我困惑很久。水位用來控制消息隊列中數據的大小，高水位（high_water_mark）用于控制消息隊列的上限，它用于控制生產者往里面放數據的量，如果消息隊列中數據量已經達到高水位，而用使用了鎖，既使用“ACE_Message_Queue_Factory<ACE_MT_SYNCH>::create_static_message_queue();”創建消息隊列，那么生產者將被阻塞。高水位很容易理解，但是低水位是用來做什么的呢？我本來地以為所謂的低水位嘛，肯定是用來控制消費者的，比如低水位為20，那么當數據量少于20時，消費者將被阻塞，直到生產者往隊列中放入數據，數據量大于20時喚醒消費者即可，消費者此時可以將低水位調低比如到0，一次將這20個單位的數據消費掉，然后再次將低水位調至20，下一次循環將自己阻塞，如此這般就可以很好地避免“傻窗口綜合癥”( silly window syndrome )。帶著這個理解去測試靜態隊列時讓我郁悶了好一陣子，看了一陣子源代碼之后才發現事實并不是這么回事。只要消息隊列中還有數據消費者就不會被阻塞的，而當數據量超過高水位時，生產者會被阻塞，既然會被阻塞，那么它肯定需要被喚醒，那么什么時候由誰來喚醒生產者呢？這就是低水位的作用，消費者一直消費數據，當數據低于低水位時它就喚醒生產者。我感覺這似乎并不能解決傻窗口綜合癥，可能我是誤解了傻窗口綜合癥吧。

        下面的代碼很好的展示了靜態消息隊列的使用。

#include "ace/Message_Queue.h"
#include "ace/Get_Opt.h"
#include "ace/OS.h"
#include  <ace/Thread_Manager.h>
#include <ace/Synch.h>

//消息隊列指針
ACE_Message_Queue<ACE_MT_SYNCH>* mq;

const char S1[] = "C++";
const char S2[] = "Java";
const char S3[] = "PHP";
const char S4[] = "C#";

//四個消息指針
ACE_Message_Block* mb1, * mb2, * mb3, * mb4;

//生產者
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)
   {
    //將高水位設置為10， S1+S2的長度為3+4+2=9<10，因此可以將S3放進去
    //但是再放入S4時生產者將會被阻塞
    //需要注意的是水位的大小并不是消息的個數，而是消息隊列中消息里面的數據量之和
    //如果也能以消息的個數作為高低水位的值就好了
    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"));

    //因為消費者在睡眠6秒之后才會調用deactivate，因此生產者會在這兒阻塞幾秒鐘
    //可以不斷地將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);

    //將低水位設置為5，因為高水位仍然為10，當前的數據量又超過了10，
    //所以下面的入隊操作仍會將生產者阻塞
    //這樣消費者消費消息，當數據量小于5時，將喚醒生產者
    //生產者在此處等待被消費者喚醒
    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;
}

//消費者
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秒，此時生產者和消費者都將被阻塞
   ACE_OS::sleep(6);

   //deactivate會喚醒所有的線程，將消息隊列設置為不可用
   //以后所存取操作都會返回-1
   //這個操作會喚醒生產者
   mq->deactivate();

   ++loop;
  }

  if(loop == 2)
  {
   ACE_OS::sleep(2);

   //將消息隊列的狀態設置成ACTIVATED
   //消息又可以使用了
   mq->activate();

   ++loop;
  }

  if(loop == 3)
  {
   ACE_OS::sleep(10);

   //消費兩個消息之后，數據量就小于5了，低于低水位將喚醒生產者
   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;

 //使用隨機數作為消息的優先級
 //數字越高，優先級越高
 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);

 //將消息壓入隊列中，enqueue_prio根據消息的優先級將消息放到適當的位置上
 //enqueue_head只是簡單地將數據存入隊列中，而不考慮消息的優先級
 //使用enqueue_prio壓入消息后，可以簡單通過dequeue_head和dequeue_tail
 //分別按優先級從高到低和從低到高取消息
 //如果使用enqueue_head和enqueue_tail壓入消息
 //則需要通過dequeue_prio來按照消息的優先級依次將消息出隊列
 //沒有必要既使用enqueue_prio壓入消息，又實用dequeue_prio來取消息
 mq->enqueue_prio (mb1);
 mq->enqueue_prio (mb2);
 mq->enqueue_prio (mb3);
 mq->enqueue_prio (mb4);

 //輸出靜態消息隊列的相關信息
 //高低水位默認值均為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遍歷消息，遍歷的順序為高優先級到底優先級
 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());

 //使用迭代器遍歷消息隊列，遍歷的順序為高優先級到底優先級
 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()));

  //這里如果不判斷的話，消息隊列空時會導致主線程被阻塞
  if(mq->is_empty())
   break;
 }

 ACE_DEBUG((LM_DEBUG, "\n\n"));

 //////////////////////////////////測試高低水位和隊列的state使用，進行測試之前mq隊列已空///////////////////////////
 //產生一個生產者線程
 ACE_Thread_Manager::instance()->spawn_n
         (
          1,
          (ACE_THR_FUNC) produce
    );

 ////產生兩個消費者線程
 ACE_Thread_Manager::instance()->spawn_n
         (
          1,
          (ACE_THR_FUNC) consume
    );

 //掛起主線程
 ACE_Thread_Manager::instance()->wait();
 
 return 0;
}

轉自http://iknow.4ucode.com/Study/Topic/895400