如何用Java获取Apache Camel集成的ActiveMQ队列消息处理量？

你这边用Apache Camel搭配ActiveMQ实现了消息队列的发布订阅，现在想通过独立Java程序拿到队列的消息发布、订阅数量，但尝试JMX和Jolokia都碰到了连接问题，我来帮你一步步解决这些问题，同时提供可用的示例代码。

一、先搞定JMX连接失败的问题 #

你遇到的NameNotFoundException: jmxrmi，主要是两个原因：要么是ActiveMQ的JMX配置有问题，要么是你用的ObjectName路径不对。

1. 确认ActiveMQ的JMX配置没问题 #

打开ActiveMQ的activemq.xml，找到managementContext节点，确保它是这样配置的（如果没有就加上）：

<managementContext>
    <managementContext createConnector="true" connectorPort="1099" connectorPath="/jmxrmi" bindAddress="0.0.0.0"/>
</managementContext>

这里要注意：

• bindAddress设为0.0.0.0才能允许远程机器访问，要是设成localhost就只能本地连
• 防火墙一定要开放1099端口，不然远程连不上

2. 修正你的ObjectName格式 #

你的ObjectName可能写错了，正确的格式应该是：
org.apache.activemq:type=Broker,brokerName=<你的Broker名称>,destinationType=Queue,destinationName=<队列名>
怕写错的话，你可以先用JConsole连接到ActiveMQ的JMX服务，直接查看对应队列的MBean路径，这样绝对不会错。

3. 可用的JMX示例代码 #

我给你调整了代码，加上了异常处理和正确的连接逻辑，你可以直接用：

import javax.management.MBeanServerConnection;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import org.apache.activemq.broker.jmx.DestinationViewMBean;

public class ActiveMQJMXStats {
    public static void main(String[] args) {
        String jmxUrl = "service:jmx:rmi:///jndi/rmi://abcdefdv302.attpmerd.com:1099/jmxrmi";
        String brokerName = "abcdefdv302.attpmerd.com";
        String queueName = "tmp.incoming.master";

        try {
            JMXServiceURL serviceURL = new JMXServiceURL(jmxUrl);
            JMXConnector connector = JMXConnectorFactory.connect(serviceURL);
            MBeanServerConnection connection = connector.getMBeanServerConnection();

            // 构建正确的队列MBean名称
            ObjectName queueMBeanName = new ObjectName(
                String.format("org.apache.activemq:type=Broker,brokerName=%s,destinationType=Queue,destinationName=%s",
                brokerName, queueName)
            );

            // 获取队列的MBean代理
            DestinationViewMBean queueMBean = javax.management.remote.rmi.RMIConnectorServer.newProxyInstance(
                connection, queueMBeanName, DestinationViewMBean.class, true
            );

            // 读取你需要的统计数据
            long queueSize = queueMBean.getQueueSize();
            long totalPublished = queueMBean.getEnqueueCount(); // 总发布消息数
            long totalConsumed = queueMBean.getDequeueCount(); // 总消费消息数
            long currentConsumers = queueMBean.getConsumerCount(); // 当前在线消费者数

            System.out.println("当前队列消息数: " + queueSize);
            System.out.println("累计发布消息数: " + totalPublished);
            System.out.println("累计消费消息数: " + totalConsumed);
            System.out.println("当前消费者数量: " + currentConsumers);

            connector.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

二、解决Jolokia连接被拒绝的问题 #

你碰到的Connection refused，大概率是这几个原因：

• ActiveMQ没启用Jolokia插件
• 8161端口被防火墙挡住了
• 地址或者端口写错了

1. 确认ActiveMQ启用了Jolokia #

打开ActiveMQ的conf/jetty.xml，检查有没有Jolokia的Servlet配置，如果没有就加上：

<bean id="jolokia" class="org.eclipse.jetty.servlet.ServletHolder">
    <property name="servlet">
        <bean class="org.jolokia.http.AgentServlet"/>
    </property>
</bean>
<bean id="jolokiaMapping" class="org.eclipse.jetty.servlet.ServletMapping">
    <property name="servletName" value="jolokia"/>
    <property name="pathSpec" value="/api/jolokia/*"/>
</bean>

同时确认jetty.xml里的端口是8161，并且绑定地址是0.0.0.0，允许远程访问。

2. 可用的Jolokia示例代码 #

下面是调整后的Jolokia代码，能正确获取队列统计：

import org.jolokia.client.J4pClient;
import org.jolokia.client.request.J4pReadRequest;
import org.jolokia.client.request.J4pReadResponse;

public class ActiveMQJolokiaStats {
    public static void main(String[] args) {
        String jolokiaUrl = "http://atlesbdv02.amd.com:8161/api/jolokia";
        String brokerName = "atlesbdv02.amd.com";
        String queueName = "tmp.incoming.master";

        try {
            J4pClient j4pClient = new J4pClient(jolokiaUrl);

            // 构建要读取的MBean路径和属性
            String mBeanPath = String.format("org.apache.activemq:type=Broker,brokerName=%s,destinationType=Queue,destinationName=%s",
                    brokerName, queueName);
            J4pReadRequest request = new J4pReadRequest(mBeanPath, "QueueSize", "EnqueueCount", "DequeueCount", "ConsumerCount");

            J4pReadResponse response = j4pClient.execute(request);

            // 解析返回的统计数据
            long queueSize = (Long) response.getValue("QueueSize");
            long totalPublished = (Long) response.getValue("EnqueueCount");
            long totalConsumed = (Long) response.getValue("DequeueCount");
            long currentConsumers = (Long) response.getValue("ConsumerCount");

            System.out.println("当前队列消息数: " + queueSize);
            System.out.println("累计发布消息数: " + totalPublished);
            System.out.println("累计消费消息数: " + totalConsumed);
            System.out.println("当前消费者数量: " + currentConsumers);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

三、需要的依赖 #

不管用JMX还是Jolokia，都要在你的项目里加对应的Maven依赖（版本换成你用的ActiveMQ版本）：

<!-- ActiveMQ JMX相关依赖 -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-broker</artifactId>
    <version>5.16.5</version>
</dependency>

<!-- Jolokia客户端依赖 -->
<dependency>
    <groupId>org.jolokia</groupId>
    <artifactId>jolokia-client-java</artifactId>
    <version>1.7.1</version>
</dependency>

四、排查小技巧 #

1. 先测网络连通性：用telnet abcdefdv302.attpmerd.com 1099或者nc -zv abcdefdv302.attpmerd.com 1099测试端口能不能通，不通的话先搞网络/防火墙
2. 用JConsole验证JMX：本地打开JConsole，输入你的JMX URL，看看能不能看到ActiveMQ的MBeans，顺便确认ObjectName是不是正确的
3. 看ActiveMQ日志：打开data/activemq.log，看看JMX和Jolokia服务有没有正常启动，有没有报错信息

内容的提问来源于stack exchange，提问作者Santosh Kumar