首页 > 代码库 > (转)ActiveMQ消息特性:Advisory Message(通知消息)

(转)ActiveMQ消息特性:Advisory Message(通知消息)

简单的说就是实现了ActiveMQ的broker上各种操作的记录跟踪和通知。

 

使用这个功能,你可以实时的知道broker上

 

  1. 创建或销毁了连接,
  2. 添加或删除了生存者或消费者,
  3. 添加或删除了主题或队列,
  4. 有消息发送和接收,
  5. 什么时候有慢消费者,
  6. 什么时候有快生产者
  7. 什么时候什么消息被丢弃
  8. 什么时候broker被添加到集群(主从或是网络连接)

 

这个机制是ActiveMQ对JMS协议的重要补充,也是基于JMS实现的ActiveMQ的可管理性的一部分。多个ActiveMQ的相互协调和互操作的基础设置。

 

使用示例与注意事项


使用很简单,完全的jms方式,示例如下

 

 

...    Destination advisoryDestination = AdvisorySupport.getProducerAdvisoryTopic(destination)    MessageConsumer consumer = session.createConsumer(advisoryDestination);    consumer.setMessageListener(this);....public void onMessage(Message msg){    if (msg instanceof ActiveMQMessage){        try {             ActiveMQMessage aMsg =  (ActiveMQMessage)msg;             ProducerInfo prod = (ProducerInfo) aMsg.getDataStructure();        } catch (JMSException e) {            log.error("Failed to process message: " + msg);        }    }}

通过拿到的消息的数据结构,可以进一步的拿到相关操作信息。

 

而且Advisory消息都是先返回当前的所有存在数据,比如现在已经存在的所有连接或是队列信息,然后再实时通知你新创建和断开的连接,新添加或删除的队列等等。。。

ActiveMQ.Advisory.Queue 数据结构为DestinationInfo,先拿到broker上现有的所有队列列表
如果再有add或remove队列的操作,拿到通知,operationType=0为add,为1则为remove
为topic时,自动过滤掉Advisory的topic。

ActiveMQ.Advisory.Producer.Queue 数据结构为ProducerInfo,其中包含Producer的参数信息
ActiveMQ.Advisory.Consumer.Queue 数据结构为ConsumerInfo,其中包含Consumer的参数信息
使用的时候,需要以此为前缀,监听需要的队列,比如kk.adt,则应该写
Topic topic = new ActiveMQTopic("ActiveMQ.Advisory.Consumer.Queue.kk.adt");
监听所有队列则
Topic topic = new ActiveMQTopic("ActiveMQ.Advisory.Consumer.Queue..>");
Producer或Consumer断开时,数据结构为RemoveInfo。
主题的监听类似。

持久订阅者上线时是ConsumerInfo,里面有clientId和consumerId,下线时的RemoveInfo里有consumerId,跟上线时对应。

ActiveMQ内置了一些工具类,根据你需要监听的队列名,自动拼Advisory的名称:
例如 AdvisorySupport.getProducerAdvisoryTopic(new ActiveMQTopic("kk.adt"));

broker接收到或是投递消息的通知默认是关闭的。
需要在policy上设置相应的开关。 消息处理相关的Advisory的数据结构是消息本身。

客户端连接相关的通知消息 都是默认开启的

Advisory TopicsDescriptionpropertiesData Structure
ActiveMQ.Advisory.ConnectionConnection start & stop messages
ActiveMQ.Advisory.Producer.QueueProducer start & stop messages on a QueueString=‘producerCount‘ - the number of producersProducerInfo
ActiveMQ.Advisory.Producer.TopicProducer start & stop messages on a TopicString=‘producerCount‘ - the number of producersProducerInfo
ActiveMQ.Advisory.Consumer.QueueConsumer start & stop messages on a QueueString=‘consumerCount‘ - the number of ConsumersConsumerInfo
ActiveMQ.Advisory.Consumer.TopicConsumer start & stop messages on a TopicString=‘consumerCount‘ - the number of ConsumersConsumerInfo

 

队列和消息的通知消息  队列相关的是默认开启的,消息相关的都需要手工配置Pilocy选项开启

Advisory TopicsDescriptionpropertiesData StructuredefaultPolicyEntry property
ActiveMQ.Advisory.QueueQueue create & destroynullnulltruenone
ActiveMQ.Advisory.TopicTopic create & destroynullnulltruenone
ActiveMQ.Advisory.TempQueueTemporary Queue create & destroynullnulltruenone
ActiveMQ.Advisory.TempTopicTemporary Topic create & destroynullnulltruenone
ActiveMQ.Advisory.Expired.QueueExpired messages on a QueueString=‘orignalMessageId‘ - the expired idMessagetruenone
ActiveMQ.Advisory.Expired.TopicExpired messages on a TopicString=‘orignalMessageId‘ - the expired idMessagetruenone
ActiveMQ.Advisory.NoConsumer.QueueNo consumer is available to process messages being sent on a QueuenullMessagefalsesendAdvisoryIfNoConsumers
ActiveMQ.Advisory.NoConsumer.TopicNo consumer is available to process messages being sent on a TopicnullMessagefalsesendAdvisoryIfNoConsumers

从5.2版本新加的消息处理与性能相关的,都需要手工启动

Advisory TopicsDescriptionpropertiesData StructuredefaultPolicyEntry property
ActiveMQ.Advisory.SlowConsumer.QueueSlow Queue ConsumerString=‘consumerId‘ - the consumer idConsumerInfofalseadvisoryForSlowConsumers
ActiveMQ.Advisory.SlowConsumer.TopicSlow Topic ConsumerString=‘consumerId‘ - the consumer idConsumerInfofalseadvisoryForSlowConsumers
ActiveMQ.Advisory.FastProducer.QueueFast Queue producerString=‘producerId‘ - the producer idProducerInfofalseadvisdoryForFastProducers
ActiveMQ.Advisory. FastProducer.TopicFast Topic producerString=‘consumerId‘ - the producer idProducerInfofalseadvisdoryForFastProducers
ActiveMQ.Advisory.MessageDiscarded.QueueMessage discardedString=‘orignalMessageId‘ - the discarded idMessagefalseadvisoryForDiscardingMessages
ActiveMQ.Advisory.MessageDiscarded.TopicMessage discardedString=‘orignalMessageId‘ - the discarded idMessagefalseadvisoryForDiscardingMessages
ActiveMQ.Advisory.MessageDelivered.QueueMessage delivered to the brokerString=‘orignalMessageId‘ - the delivered idMessagefalseadvisoryForDelivery
ActiveMQ.Advisory.MessageDelivered.TopicMessage delivered to the brokerString=‘orignalMessageId‘ - the delivered idMessagefalseadvisoryForDelivery
ActiveMQ.Advisory.MessageConsumed.QueueMessage consumed by a clientString=‘orignalMessageId‘ - the delivered idMessagefalseadvisoryForConsumed
ActiveMQ.Advisory.MessageConsumed.TopicMessage consumed by a clientString=‘orignalMessageId‘ - the delivered idMessagefalseadvisoryForConsumed
ActiveMQ.Advisory.FULLA Usage resource is at its limitString=‘usageName‘ - the name of Usage resourcenullfalseadvisoryWhenFull
ActiveMQ.Advisory.MasterBrokerA broker is now the master in a master/slave configurationnullnulltruenone

从5.4版本新加的DLQ的通知消息

Advisory TopicsDescriptionpropertiesData StructuredefaultPolicyEntry property
ActiveMQ.Advisory.MessageDLQd.QueueMessage sent to DLQString=‘orignalMessageId‘ - the delivered idMessageAlways onadvisoryForConsumed
ActiveMQ.Advisory.MessageDLQd.TopicMessage sent to DLQString=‘orignalMessageId‘ - the delivered idMessageAlways onadvisoryForConsumed

从5.5版本新加的网络连接集群的通知消息

Advisory TopicsDescriptionpropertiesData Structuredefault
ActiveMQ.Advisory.NetworkBridgeNetwork bridge being stopped or startedBoolean="started" - true if bridge is started, false if it is stopped
Boolean="createdByDuplex" - true if the bridge is created by remote network connector
BrokerInfo - provides data of the remote brokerAlways on

打开通知消息开关的示例

<destinationPolicy>   <policyMap><policyEntries>       <policyEntry topic=">" advisoryForConsumed="true" />   </policyEntries></policyMap></destinationPolicy>

 

此外,在broker上有个通知消息的总开关,设置false以后,所有的Advisory都不可用:

 

 <broker advisorySupport="false">...
转自:http://blog.csdn.net/kimmking/article/details/8443679

 

(转)ActiveMQ消息特性:Advisory Message(通知消息)