ActiveMQ连接事件通知：利用Advisory Topics监控代理状态

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMessage; // 用于访问ActiveMQ特有的消息属性

import javax.jms.*;

public class ActiveMQConnectionMonitor {

    // ActiveMQ代理的URL
    private static final String BROKER_URL = "tcp://localhost:61616";
    // 监听连接事件的Advisory Topic名称
    private static final String ADVISORY_CONNECTION_TOPIC = "ActiveMQ.Advisory.Connection";

    public static void main(String[] args) {
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;

        try {
            // 1. 创建JMS连接工厂
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);

            // 2. 创建并启动JMS连接
            connection = connectionFactory.createConnection();
            connection.start();

            // 3. 创建JMS会话 (非事务性，自动确认消息)
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 4. 创建Advisory Topic对象
            Topic advisoryTopic = session.createTopic(ADVISORY_CONNECTION_TOPIC);

            // 5. 创建消息消费者，用于订阅Advisory Topic
            consumer = session.createConsumer(advisoryTopic);

            // 6. 设置消息监听器，当收到Advisory消息时进行处理
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        // Advisory消息通常是ActiveMQMessage类型，包含特定的属性
                        if (message instanceof ActiveMQMessage) {
                            ActiveMQMessage amqMessage = (ActiveMQMessage) message;

                            // 从消息属性中提取事件信息
                            String advisoryMessageType = amqMessage.getStringProperty("AdvisoryMessageType");
                            String connectionId = amqMessage.getStringProperty("connectionId");
                            boolean isConnectionStart = amqMessage.getBooleanProperty("isConnectionStart");
                            boolean isConnectionStop = amqMessage.getBooleanProperty("isConnectionStop");

                            System.out.println("----------------------------------------");
                            System.out.println("收到ActiveMQ连接事件通知：");
                            System.out.println("  消息ID: " + amqMessage.getJMSMessageID());
                            System.out.println("  Advisory消息类型: " + advisoryMessageType);
                            System.out.println("  关联连接ID: " + connectionId);
                            System.out.println("  是连接启动事件? " + isConnectionStart);
                            System.out.println("  是连接停止事件? " + isConnectionStop);
                            System.out.println("----------------------------------------");

                            // 根据事件类型执行相应的业务逻辑
                            if (isConnectionStart) {
                                System.out.println(" -> 新连接已建立: " + connectionId);
                                // 例如：记录日志、更新连接状态仪表盘等
                            } else if (isConnectionStop) {
                                System.out.println(" -> 连接已关闭: " + connectionId);
                                // 例如：清理资源、发送告警等
                            }

                        } else {
                            System.out.println("收到非ActiveMQMessage类型消息: " + message.getClass().getName());
                        }
                    } catch (JMSException e) {
                        System.err.println("处理Advisory消息时发生错误: " + e.getMessage());
                        e.printStackTrace();
                    }
                }
            });

            System.out.println("正在监听ActiveMQ连接事件... 请启动或关闭其他ActiveMQ客户端进行测试。");
            System.out.println("按Ctrl+C或关闭程序退出监听。");

            // 保持主线程运行，以便监听器可以持续接收消息
            Thread.sleep(Long.MAX_VALUE);

        } catch (JMSException e) {
            System.err.println("JMS操作失败: " + e.getMessage());
            e.printStackTrace();
        } catch (InterruptedException e) {
            System.err.println("监听线程中断: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 7. 关闭JMS资源，释放连接
            try {
                if (consumer != null) consumer.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
                System.out.println("JMS资源已关闭。");
            } catch (JMSException e) {
                System.err.println("关闭JMS资源失败: " + e.getMessage());
                e.printStackTrace();
            }
        }
    }
}

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.18.3</version> <!-- 使用您ActiveMQ版本对应的客户端版本 -->
</dependency>