ACE之ACE_Message

ACE之ACE_Message_Queue

news2025/1/4 2:25:18

简介

ACE_Message_Queue是ACE中通用消息队列，其是ACE_Task以及并发中ACE_Activation_Queue的基础

结构

ACE_Message_Queue_Base

ACE_Message_Queue_Base 是消息队列的抽象类，与其相关的枚举有队列的状态以及高位，低位大小

enum
 {
   DEFAULT_HWM = 16 * 1024,
   DEFAULT_LWM = 16 * 1024,

   ACTIVATED = 1,
   DEACTIVATED = 2,
   PULSED = 3
 };

包含一个表示状态的成员

protected:
  /// <ACTIVATED>, <DEACTIVATED>, or <PULSED>.
  int state_;

相关的抽象方法

virtual int peek_dequeue_head (ACE_Message_Block *&first_item,
                                 ACE_Time_Value *timeout = 0) = 0;
virtual int enqueue_tail (ACE_Message_Block *new_item,
                          ACE_Time_Value *timeout = 0) = 0;
virtual int enqueue (ACE_Message_Block *new_item,
                     ACE_Time_Value *timeout = 0) = 0;
virtual int dequeue_head (ACE_Message_Block *&first_item,
                          ACE_Time_Value *timeout = 0) = 0;
virtual int dequeue (ACE_Message_Block *&first_item,
                     ACE_Time_Value *timeout = 0) = 0;
virtual bool is_full (void) = 0;
virtual bool is_empty (void) = 0;
virtual size_t message_bytes (void) = 0;
virtual size_t message_length (void) = 0;
virtual size_t message_count (void) = 0;
virtual void message_bytes (size_t new_size) = 0;
virtual void message_length (size_t new_length) = 0;
virtual int deactivate (void) = 0;
virtual int activate (void) = 0;
virtual int pulse (void) = 0;
virtual int state (void);
virtual int deactivated (void) = 0;
virtual ACE_Notification_Strategy *notification_strategy (void) = 0;
virtual void notification_strategy (ACE_Notification_Strategy *s) = 0;
virtual void dump (void) const = 0;

队列实现类

ACE_Message_Queue是队列的实现类模板，支持同步策略和时间策略，通过ACE_Message_Block实现队列

ACE_Message_Queue_Ex：消息类型为类的第一个模板参数ACE_MESSAGE_TYPE
ACE_Message_Queue_Ex_N：在入队列前，将多个ACE_MESSAGE_TYPE拼接成一个由ACE_Message_Block构成的链表再放入队列，要求ACE_MESSAGE_TYPE类有next()方法

template <class ACE_MESSAGE_TYPE, ACE_SYNCH_DECL, class TIME_POLICY> ACE_Message_Block *
ACE_Message_Queue_Ex_N<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::wrap_with_mbs_i
  (ACE_MESSAGE_TYPE *new_item)
{
  ACE_TRACE ("ACE_Message_Queue_Ex_N<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::wrap_with_mbs_i");

  // We need to keep a reference to the head of the chain
  ACE_Message_Block *mb_head = 0;

  ACE_NEW_RETURN (mb_head,
                  ACE_Message_Block ((char *) new_item,
                                     sizeof (*new_item),
                                     ACE_Message_Queue_Ex<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::DEFAULT_PRIORITY),
                  0);

  // mb_tail will point to the last ACE_Message_Block
  ACE_Message_Block *mb_tail = mb_head;

  // Run through rest of the messages and wrap them
  for (ACE_MESSAGE_TYPE *pobj = new_item->next (); pobj; pobj = pobj->next ())
    {
      ACE_Message_Block *mb_temp = 0;
      ACE_NEW_NORETURN (mb_temp,
                        ACE_Message_Block ((char *) pobj,
                                           sizeof (*pobj),
                                           ACE_Message_Queue_Ex<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::DEFAULT_PRIORITY));
      if (mb_temp == 0)
        {
          mb_head->release ();
          mb_head = 0;
          break;
        }

      mb_tail->next (mb_temp);
      mb_tail = mb_temp;
    }

  return mb_head;
}