Activemq获取消息信息

Activemq的公告消息

下面是来自官网的翻译：http://activemq.apache.org/advisory-message.html

ActiveMQ消息属性

消息属性

属性名	类型	默认值	描述
JMSDestination	javax.jms.Destination	生产者set进去	发送消息目的地
JMSReplyTo	javax.jms.Destination	null	用户定义
JMSType	String	“”	用户定义
JMSDeliveryMode	int	DeliveryMode.PERSISTENT	消息是否要持久化
JMSPriority	int	4	0-9
JMSMessageID	String	unique	消息唯一标示符
JMSTimestamp	long	消息发送时间	毫秒
JMSCorrelationID	String	null	用户定义
JMSExpiration	long	0	0表示用不过气，消息过期时间
JMSRedelivered	boolean	false	重新传递

JMS定义属性

属性名	类型	默认值	描述
JMSXDeliveryCount	int	0	尝试发送消息的次数
JMSXGroupID	String	null	表示消息的分组
JMSXGroupSeq	int	0	消息的序列号
JMSXProducerTXID	String	null	事务标示

ActiveMQ定义属性

属性名	类型	默认值	描述
JMSActiveMQBrokerInTime	long	0	消息到达broker的时间
JMSActiveMQBrokerOutTime	long	0	消息离开broker的时间

公告消息（Advisory Message）

ActiveMQ支持的公告消息（相当于监听器）

• 消费者，生产者和connection开始和停止

• 临时目的地的创建和销毁

• 话题和队列上的消息过期

• broker发送消息到没有消费者的目的地

• connections开启和停止

  公告消息是一种管理JMS的渠道，我们可以接收到在broker上的生产者消费者目的地的变化。
  当我们通过JMX查看broker时候，可以看到公告话题以ActiveMQ.Advisory为前缀，每个公告拥有类型为Advisory的消息和一些预定义的消息属性

属性名	类型	描述
originBrokerId	StringProperty	公告起源的broker的ID
originBrokerName	StringProperty	公告起源的broker的名称
originBrokerURL	StringProperty	公告起源的第一个broker url

我们可以获得到消息的一些元数据

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

支持的公告话题

客户端的公告消息
我们可以用consumerCount头信息获取当前消费者的数量来得知当前的公告的消息是否已经发送。

Advisory Topics	Description	properties	Data Structure
ActiveMQ.Advisory.Connection	ActiveMQ.Advisory.Connection		ConnectionInfo, RemoveInfo
ActiveMQ.Advisory.Producer.Queue	Producer start & stop messages on a Queue	String=’producerCount’ - the number of producers	ProducerInfo
ActiveMQ.Advisory.Producer.Topic	Producer start & stop messages on a Topic	String=’producerCount’ - the number of producers	ProducerInfo
ActiveMQ.Advisory.Consumer.Queue	Consumer start & stop messages on a Queue	String=’consumerCount’ - the number of Consumers	ConsumerInfo, RemoveInfo
ActiveMQ.Advisory.Consumer.Topic	Consumer start & stop messages on a Topic	String=’consumerCount’ - the number of Consumers	ConsumerInfo, RemoveInfo

目的地和消息的公告

Advisory Topics	描述	属性	数据结构	默认值	策略属性
ActiveMQ.Advisory.Queue	Queue create & destroy	null	DestinationInfo	true	none
ActiveMQ.Advisory.Topic	Topic create & destroy	null	DestinationInfo	true	none
ActiveMQ.Advisory.TempQueue	Temporary Queue create & destroy	null	DestinationInfo	true	none
ActiveMQ.Advisory.TempTopic	Temporary Topic create & destroy	null	DestinationInfo	true	none
ActiveMQ.Advisory.Expired.Queue	Expired messages on a Queue	String=’orignalMessageId’ - the expired id	Message	true	none
ActiveMQ.Advisory.Expired.Topic	Expired messages on a Topic	String=’orignalMessageId’ - the expired id	Message	true	none
ActiveMQ.Advisory.NoConsumer.Queue	No consumer is available to process messages being sent on a Queue	null	Message	false	sendAdvisoryIfNoConsumers
ActiveMQ.Advisory.NoConsumer.Topic	No consumer is available to process messages being sent on a Topic	null	Message	false	sendAdvisoryIfNoConsumers

ActiveMQ5.2新的公告消息

Advisory Topics	描述	属性	数据结构	默认值	策略属性
ActiveMQ.Advisory.SlowConsumer.Queue	Slow Queue Consumer	String=’consumerId’ - the consumer id	ConsumerInfo	false	advisoryForSlowConsumers
ActiveMQ.Advisory.SlowConsumer.Topic	Slow Topic Consumer	String=’consumerId’ - the consumer id	ConsumerInfo	false	advisoryForSlowConsumers
ActiveMQ.Advisory.FastProducer.Queue	Fast Queue producer	String=’producerId’ - the producer id	ProducerInfo	false	advisdoryForFastProducers
ActiveMQ.Advisory. FastProducer.Topic	Fast Topic producer	String=’consumerId’ - the producer id	ProducerInfo	false	advisdoryForFastProducers
ActiveMQ.Advisory.MessageDiscarded.Queue	Message discarded	String=’orignalMessageId’ - the discarded id	Message	false	advisoryForDiscardingMessages
ActiveMQ.Advisory.MessageDiscarded.Topic	Message discarded	String=’orignalMessageId’ - the discarded id	Message	false	advisoryForDiscardingMessages
ActiveMQ.Advisory.MessageDelivered.Queue	Message delivered to the broker	String=’orignalMessageId’ - the delivered id	Message	false	advisoryForDelivery
ActiveMQ.Advisory.MessageDelivered.Topic	Message delivered to the broker	String=’orignalMessageId’ - the delivered id	Message	false	advisoryForDelivery
ActiveMQ.Advisory.MessageConsumed.Queue	Message consumed by a client	String=’orignalMessageId’ - the delivered id	Message	false	advisoryForConsumed
ActiveMQ.Advisory.MessageConsumed.Topic	Message consumed by a client	String=’orignalMessageId’ - the delivered id	Message	false	advisoryForConsumed
ActiveMQ.Advisory.FULL	A Usage resource is at its limit	String=’usageName’ - the name of Usage resource	null	false	advisoryWhenFull
ActiveMQ.Advisory.MasterBroker	A broker is now the master in a master/slave configuration	null	null	true	none

5.4后新公告

Advisory Topics	描述	属性	数据结构	默认值	策略属性
ActiveMQ.Advisory.MessageDLQd.Queue	Message sent to DLQ	String=’orignalMessageId’ - the delivered id	Message	Always on	advisoryForConsumed
ActiveMQ.Advisory.MessageDLQd.Topic	Message sent to DLQ	String=’orignalMessageId’ - the delivered id	Message	Always on	advisoryForConsumed

Network bridge advisories

Advisory Topics	描述	属性	数据结构	默认值
ActiveMQ.Advisory.NetworkBridge	Network bridge being stopped or started	Boolean=”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 broker	Always on

开启公告消息

公告消息默认是不开启的，我们可以使用下面的配置开启公告配置

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

关闭公告消息

xml配置如下

<broker advisorySupport="false">

java code 配置

BrokerService broker = new BrokerService();
broker.setAdvisorySupport(false);
broker.start();

url链接串

"tcp://localhost:61616?jms.watchTopicAdvisories=false"

设置ConnectionFactory的属性

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
factory.setWatchTopicAdvisories(false);

ActiveMQ一些静态方法

AdvisorySupport.getConsumerAdvisoryTopic()
AdvisorySupport.getProducerAdvisoryTopic()
AdvisorySupport.getExpiredTopicMessageAdvisoryTopic()
AdvisorySupport.getExpiredQueueMessageAdvisoryTopic()
AdvisorySupport.getNoTopicConsumersAdvisoryTopic()
AdvisorySupport.getNoQueueConsumersAdvisoryTopic()
AdvisorySupport.getDestinationAdvisoryTopic()
AdvisorySupport.getExpiredQueueMessageAdvisoryTopic()
AdvisorySupport.getExpiredTopicMessageAdvisoryTopic()
AdvisorySupport.getNoQueueConsumersAdvisoryTopic()
AdvisorySupport.getNoTopicConsumersAdvisoryTopic()

//Version 5.2 onwards


AdvisorySupport.getSlowConsumerAdvisoryTopic()
AdvisorySupport.getFastProducerAdvisoryTopic()
AdvisorySupport.getMessageDiscardedAdvisoryTopic()
AdvisorySupport.getMessageDeliveredAdvisoryTopic()
AdvisorySupport.getMessageConsumedAdvisoryTopic()
AdvisorySupport.getMasterBrokerAdvisoryTopic()
AdvisorySupport.getFullAdvisoryTopic()

具体可以看一下AdvisorySupport这个类就懂了

应用场景：

本来工作中用websocket当聊天的，结果经过F5后要做了一层代理，消息就不通了，于是就用的mq，需要捕获一个队列的消费者数量的变化，就用了Activemq的公告消息

    @Test
    public void test1(){
        Connection connection = null;
        try {
            connection = connectionFactory.createConnection();
            connection.start();
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            // 创建 advisory topic : ActiveMQ.Advisory.Consumer.Topic.chat1,用于监控topic消费者的状态变化
            ActiveMQTopic advisoryTopic = AdvisorySupport.getConsumerAdvisoryTopic(new ActiveMQTopic("topic01"));
            MessageConsumer consumer = session.createConsumer(advisoryTopic);
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    if (message instanceof ActiveMQMessage) {
                        try {
                            ActiveMQMessage aMsg = (ActiveMQMessage) message;
                            System.out.println(aMsg.getStringProperty("consumerCount"));
                            System.out.println(aMsg.getStringProperty("producerCount"));
                            if (aMsg.getDataStructure() instanceof ConsumerInfo) {
                                // Consumer start
                                ConsumerInfo consumerInfo = (ConsumerInfo) aMsg.getDataStructure();
                                System.out.println(consumerInfo);
                            } else if (aMsg.getDataStructure() instanceof RemoveInfo) {
                                // Consumer stop
                                RemoveInfo removeInfo = (RemoveInfo) aMsg.getDataStructure();
                                System.out.println(removeInfo);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            Thread.sleep(2000000);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            try {
                if(null != connection){
                    connection.close();
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

测试：先上线后下线

1
null
ConsumerInfo {commandId = 6, responseRequired = true, consumerId = ID:WGR-PC-6298-1602771436318-1:1:2:1, destination = topic://topic01, prefetchSize = 32767, maximumPendingMessageLimit = 0, browser = false, dispatchAsync = true, selector = null, clientId = wgr, subscriptionName = null, noLocal = false, exclusive = false, retroactive = false, priority = 0, brokerPath = null, optimizedAcknowledge = false, noRangeAcks = false, additionalPredicate = null}


0
null
RemoveInfo {commandId = 0, responseRequired = true, objectId = ID:WGR-PC-6298-1602771436318-1:1:2:1, lastDeliveredSequenceId = -2}

其实除了JMX，咨询消息，还有统计插件，没有深入研究，官网说明如下：

http://activemq.apache.org/statisticsplugin.html

示例：https://segmentfault.com/a/1190000004522121?utm_source=debugrun&utm_medium=referral

从ActiveMQ 5.3开始，其中包括一个统计信息插件，可用于从代理或其目的地检索统计信息。请注意，该消息必须包含replyTo标头（jmsReplyTo如果使用的是JMS，则为标头），否则该消息将被忽略。该replyTo标题必须包含从中要获取统计信息（一个或多个）目标的名称。统计消息中MapMessage填充了目标（即，代理或目标）的统计信息。

要检索代理的统计信息，只需向标ActiveMQ.Statistics.Broker有replyTo标题的目的地发送一条空消息。要检索目的地统计，只需发送一个空的消息载明的目的地ActiveMQ.Statistics.Destination.<destination-name>或ActiveMQ.Statistics.Destination.<wildcard-expression>连同replyTo头。如果许多目的地与给定的通配符表达式匹配，则将向每个目的地发送一条状态消息replyTo。

要将ActiveMQ配置为使用统计信息插件，只需将以下内容添加到ActiveMQ XML配置中：

<broker ...>
  <plugins>
    <statisticsBrokerPlugin/>
  </plugins>
</broker>

统计信息插件查找发送到特定目的地的消息。以下是使用统计信息插件从代理获取统计信息的示例：

Queue replyTo = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(replyTo);

String queueName = "ActiveMQ.Statistics.Broker";
Queue testQueue = session.createQueue(queueName);
MessageProducer producer = session.createProducer(testQueue);
Message msg = session.createMessage();
msg.setJMSReplyTo(replyTo);
producer.send(msg);

MapMessage reply = (MapMessage) consumer.receive();
assertNotNull(reply);

for (Enumeration e = reply.getMapNames();e.hasMoreElements();) {
  String name = e.nextElement().toString();
  System.out.println(name + "=" + reply.getObject(name));
}

上面代码的输出如下所示：

vm=vm://localhost  
memoryUsage=0  
storeUsage=3330  
tempPercentUsage=0  
ssl=  
openwire=tcp://localhost:50059  
brokerId=ID:bigmac-50057-1253605065511-0:0  
consumerCount=2  
brokerName=localhost  
expiredCount=0  
dispatchCount=1  
maxEnqueueTime=5.0  
storePercentUsage=0  
dequeueCount=0  
inflightCount=1  
messagesCached=0  
tempLimit=107374182400  
averageEnqueueTime=5.0  
stomp+ssl=  
memoryPercentUsage=0  
size=10  
tempUsage=0  
producerCount=1  
minEnqueueTime=5.0  
dataDirectory=/Users/rajdavies/dev/projects/activemq/activemq-core/activemq-data  
enqueueCount=10  
stomp=  
storeLimit=107374182400  
memoryLimit=67108864

同样，要查询目标的统计信息，只需将消息发送到以开头的目标名称即可ActiveMQ.Statistics.Destination。例如，要检索名称为TEST.FOO的队列的统计信息，请将空消息发送到名为的队列ActiveMQ.Statistics.Destination.TEST.FOO。下面是一个示例：

Queue replyTo = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(replyTo);

Queue testQueue = session.createQueue("TEST.FOO");
MessageProducer producer = session.createProducer(null);

String queueName = "ActiveMQ.Statistics.Destination." + testQueue.getQueueName()
Queue query = session.createQueue(queueName);

Message msg = session.createMessage();

producer.send(testQueue, msg) 
msg.setJMSReplyTo(replyTo);
producer.send(query, msg);
MapMessage reply = (MapMessage) consumer.receive();
assertNotNull(reply);
assertTrue(reply.getMapNames().hasMoreElements());
        
for (Enumeration e = reply.getMapNames();e.hasMoreElements();) {
    String name = e.nextElement().toString();
    System.err.println(name + "=" + reply.getObject(name));
}

上面代码的输出如下所示：

memoryUsage=0  
dequeueCount=0  
inflightCount=0  
messagesCached=0  
averageEnqueueTime=0.0  
destinationName=queue://TEST.FOO  
size=1  
memoryPercentUsage=0  
producerCount=0  
consumerCount=0  
minEnqueueTime=0.0  
maxEnqueueTime=0.0  
dispatchCount=0  
expiredCount=0  
enqueueCount=1  
memoryLimit=67108864

您也可以在队列名称中使用通配符。对于通配符匹配的每个目的地，这将导致单独的统计信息。确实非常方便。

订阅统计

从5.6.0开始，您还可以检索所有队列和主题订阅的统计信息。您所需要做的只是将空消息发送到ActiveMQ.Statistics. Subscription带有replyTo标题的目的地。响应将以一个或多个消息的形式出现，每个消息都包含关于Broker上恰好一个订阅的统计信息。

以下是使用统计信息插件从代理获取统计信息的示例：

Queue replyTo = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(replyTo);

String queueName = "ActiveMQ.Statistics.Subscription";
Queue testQueue = session.createQueue(queueName);
MessageProducer producer = session.createProducer(testQueue);
Message msg = session.createMessage();
msg.setJMSReplyTo(replyTo);
producer.send(msg);

MapMessage reply = (MapMessage) consumer.receive();
assertNotNull(reply);

for (Enumeration e = reply.getMapNames();e.hasMoreElements();) {
  String name = e.nextElement().toString();
  System.out.println(name + "=" + reply.getObject(name));
}

上面的代码示例输出如下所示：

selector=null  
dispatchedQueueSize=1  
maximumPendingMessageLimit=0  
exclusive=false  
connectionId=ID:dejan-bosanacs-macbook-pro-2.local-64989-1335528942875-4:1  
destinationName=Test.Queue  
clientId=ID:dejan-bosanacs-macbook-pro-2.local-64989-1335528942875-3:1  
slowConsumer=false  
prefetchSize=1000  
sessionId=1  
dequeueCounter=0  
enqueueCounter=1  
retroactive=false  
dispatchedCounter=1