ActiveMQ连接事件通知：Advisory Topics详解

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQMessage;

import javax.jms.*;

public class ActiveMQConnectionMonitor {

    // ActiveMQ代理的URL
    private static final String BROKER_URL = "tcp://localhost:61616";
    // 监听连接事件的Advisory Topic名称
    private static final String ADVISORY_TOPIC_NAME = AdvisorySupport.ADVISORY_TOPIC_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会话
            // 参数1: 是否启用事务; 参数2: 消息确认模式 (这里使用自动确认)
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 4. 获取Advisory Topic
            Topic advisoryTopic = session.createTopic(ADVISORY_TOPIC_NAME);
            System.out.println("正在监听Advisory Topic: " + ADVISORY_TOPIC_NAME);

            // 5. 创建消息消费者
            consumer = session.createConsumer(advisoryTopic);

            // 6. 设置消息监听器
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        if (message instanceof ActiveMQMessage) {
                            ActiveMQMessage advisoryMessage = (ActiveMQMessage) message;
                            System.out.println("------------------------------------");
                            System.out.println("接收到ActiveMQ连接事件通知：");

                            // Advisory消息通常包含丰富的属性来描述事件
                            // 可以通过检查这些属性来判断事件类型（连接创建/关闭）和连接详情
                            // 常见的属性包括：
                            // - advisoryMessage.getBooleanProperty(AdvisorySupport.MSG_PROPERTY_ADVISORY_STARTED_FROM_BROKER)
                            // - advisoryMessage.getStringProperty("connectionId")
                            // - advisoryMessage.getStringProperty("clientId")
                            // - advisoryMessage.getStringProperty("userName")
                            // - advisoryMessage.getOriginalDestination()

                            String connectionId = advisoryMessage.getStringProperty("connectionId");
                            String clientId = advisoryMessage.getStringProperty("clientId");
                            String userName = advisoryMessage.getStringProperty("userName");
                            Destination originalDestination = advisoryMessage.getOriginalDestination();

                            System.out.println("  原始目标: " + (originalDestination != null ? originalDestination.getPhysicalName() : "N/A"));
                            System.out.println("  连接ID: " + (connectionId != null ? connectionId : "N/A"));
                            System.out.println("  客户端ID: " + (clientId != null ? clientId : "N/A"));
                            System.out.println("  用户名称: " + (userName != null ? userName : "N/A"));

                            // 通过检查消息的原始目的地来推断事件类型
                            // 例如，连接创建事件可能来自特定的内部Topic
                            if (originalDestination != null) {
                                if (originalDestination.getPhysicalName().startsWith(AdvisorySupport.ADVISORY_TOPIC_CONNECTION_PREFIX + ".connections")) {
                                    // 更精确的判断需要检查消息体或更具体的属性
                                    // 对于连接创建/关闭，通常可以通过检查消息的DataStructure来获取ConnectionInfo
                                    System.out.println("  事件类型 (推断): 连接创建/关闭事件");
                                }
                            }
                            System.out.println("------------------------------------");

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

            System.out.println("ActiveMQ连接事件监听器已启动，等待连接事件...");
            // 保持主线程运行，以便监听器可以接收消息
            // 在生产环境中，通常会使用CountDownLatch或类似的同步机制来管理应用程序的生命周期
            Thread.sleep(Long.MAX_VALUE); // 简单地让主线程休眠，保持程序运行

        } catch (JMSException e) {
            System.err.println("JMS操作失败: " + e.getMessage());
            e.printStackTrace();
        } catch (InterruptedException e) {
            System.err.println("应用程序被中断: " + e.getMessage());
            Thread.currentThread().interrupt();
        } finally {
            // 清理JMS资源
            if (consumer != null) {
                try {
                    consumer.close();
                } catch (JMSException e) { /* 忽略关闭异常 */ }
            }
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) { /* 忽略关闭异常 */ }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) { /* 忽略关闭异常 */ }
            }
            System.out.println("所有JMS资源已关闭。");
        }
    }
}

<!-- Maven -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.16.5</version> <!-- 使用您实际的ActiveMQ版本 -->
</dependency>

// Gradle
implementation 'org.apache.activemq:activemq-client:5.16.5' // 使用您实际的ActiveMQ版本