查询 ActiveMQ 统计信息

目标：查询 ActiveMQ 统计信息(队列处理情况，处理耗时)
download ActiveMQ
参考：

 

centOS5.3 环境
[root@localhost opt]# uname -a

Linux localhost.localdomain 2.6.18-128.el5 #1 SMP Wed Jan 21 10:44:23 EST 2009 i686 athlon i386 GNU/Linux


#测试前先关闭防火墙，方便测试

[root@localhost opt]# service iptables stop

[root@localhost opt]# service iptables status

Firewall is stopped.

[root@localhost opt]# tar -zxvf apache-activemq-5.5.0-bin.tar.gz

#ActiveMQ 5.3以上 中新添加了一个统计插件，通过它可以查看 activemq 的运行情况

[root@localhost opt]# vi apache-activemq-5.5.0/conf/activemq.xml

#      内容

#

#

#

#apache-activemq-5.5.0/conf/activemq.xml 修改后的内容

<beans
  xmlns=""
  xmlns:amq=""
  xmlns:xsi=""
  xsi:schemaLocation=" /spring-beans-2.0.xsd
   /activemq-core.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.base}/conf/credentials.properties</value>
        </property>
    </bean>

    <broker xmlns="" brokerName="localhost" dataDirectory="${activemq.base}/data" destroyApplicationContextOnStop="true">
     
        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" producerFlowControl="true" memoryLimit="1mb">
                  <pendingSubscriberPolicy>
                    <vmCursor />
                  </pendingSubscriberPolicy>
                </policyEntry>
                <policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb">
                  <pendingQueuePolicy>
                    <vmQueueCursor/>
                  </pendingQueuePolicy>
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy>
 
        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>

        <persistenceAdapter>
            <kahaDB directory="${activemq.base}/data/kahadb"/>
        </persistenceAdapter>

    <plugins>
            <statisticsBrokerPlugin/>
        </plugins>
                    
        <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage limit="20 mb"/>
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="1 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="100 mb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>
        
    
        <transportConnectors>
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>
        </transportConnectors>

    </broker>

    <import resource="jetty.xml"/>
</beans>
#启动activeMQ

[root@localhost opt]# /opt/apache-activemq-5.5.0/bin/linux-x86-32/activemq start

#查看activeMQ 日志信息

[root@localhost ~]# tail -f /opt/apache-activemq-5.5.0/data/activemq.log



 

#java测试代码
#该测试代码针对spring jms 可直接使用，未使用spring jms 的 需做调整：
#spring 配置文件中 connectionFactory 和 队列
<bean id="connectionFactory"
    class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="failover:(tcp://192.168.20.59:61616)?initialReconnectDelay=100&maxReconnectAttempts=5" />
</bean>
<bean id="rantzDestination"
    class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg index="0" value="rantz.marketing.queue" />
</bean>

#测试servlet
package com.hc360.count.jms;

import java.io.IOException;
import java.util.Enumeration;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class MoniterServlet extends HttpServlet{

    protected void doGet(HttpServletRequest req, HttpServletResponse resp)
            throws ServletException, IOException {
        try {
            //spring context

            ApplicationContext applicationContext = new ClassPathXmlApplicationContext("..\\jmsContext.xml");
            //spring中取出连接工厂

            ConnectionFactory connectionFactory = (ConnectionFactory)applicationContext.getBean("connectionFactory");
            //取得连接

            Connection connection = connectionFactory.createConnection();
            //打开连接，非常重要

            connection.start();
            //创建没有事物的会话

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //创建零时返回队列

            Queue replyTo = session.createTemporaryQueue();
            
            //创建零时发送队列,该队列指定的名称为ActiveMQ.Statistics.Destination + 你的创建的队列名称

            Queue query = session.createQueue("ActiveMQ.Statistics.Destination"+((Queue)applicationContext.getBean("rantzDestination")).getQueueName());
            //指定目标对象为query的发送端对象

            MessageProducer producer = session.createProducer(query);
            //创建消息

            Message msg = session.createMessage();
            //指定返回队列

            msg.setJMSReplyTo(replyTo);
            //发送端发送消息

            producer.send(msg);
            
            //创建指定接收队列的接收端对象

            MessageConsumer consumer = session.createConsumer(replyTo);
            //接收消息

            MapMessage reply = (MapMessage) consumer.receive();
            
            for (Enumeration e = reply.getMapNames();e.hasMoreElements();) {
                String name = e.nextElement().toString();
                System.err.println(name+"="+reply.getObject(name));
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    protected void doPost(HttpServletRequest req, HttpServletResponse resp)
            throws ServletException, IOException {
        super.doGet(req, resp);
    }

}

#web.xml 添加servlet 配置
<servlet>
    <servlet-name>test</servlet-name>
    <servlet-class>com.hc360.count.jms.MoniterServlet</servlet-class>
</servlet>
<servlet-mapping>
    <servlet-name>test</servlet-name>
    <url-pattern>/moniter</url-pattern>
</servlet-mapping>

 

#以下为 服务器返回的统计信息
//内存使用情况
//   memoryUsage=11060
//   memoryPercentUsage=1
//   memoryLimit=1048576
   
   //The number of messages that have been acknowledged from the destination
//   dequeueCount=0
   //The number of messages dispatched but awaiting acknowledgement
//   inflightCount=0
   //The number of messages that are held in the destination's memory cache
//   messagesCached=0
   //information around length of time messages are held by a destination(平均)
//   averageEnqueueTime=0.0
   //目标队列名称
//   destinationName=queue://rantz.marketing.queue 
   //等待处理的个数
//   size=10
   //The number of producers that that are publishing messages to the destination
//   producerCount=0
   //The number of consumers that that are subscribing to messages from the destination
//   consumerCount=0
   //information around length of time messages are held by a destination(最小)
//   minEnqueueTime=0.0
   //information around length of time messages are held by a destination(最大)
//   maxEnqueueTime=0.0
   //The number of messages that have been dispatched from the destination
//   dispatchCount=0
   //The number of messages that have expired
//   expiredCount=0
   //The number of messages that have been sent to the destination
//   enqueueCount=3

#end
参考：