当生产者的连接断开时如何得到通知？

Question

我正在使用嵌入式 ActiveMQ 代理。我的目标是找到一种方法来检测队列中的外部生产者何时失去连接。

我是这样启动代理的：

BrokerService broker = new BrokerService();
broker.addConnector("tcp://" + LISTEN_DEVICE_IP + ":" + port);
setLastMessagesPersistent(broker);
broker.start();

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
connection = factory.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

connection.start();

然后我尝试添加一个 TransportListener：

((ActiveMQConnection) connection).addTransportListener(new TransportListener() {
    public void transportResumed() {
        System.out.println("resumed");
    }
    public void transportInterupted() {
        System.out.println("interrupted");
    }
    public void onException(IOException arg0) {
        System.out.println("ioexception: " + arg0);
    }
    public void onCommand(Object arg0) {
        System.out.println("command: " + arg0);
    }
});

我也像这样注册了一个consumer和ProducerListener：

Destination dest = session.createQueue(queuename);
MessageConsumer consumer = session.createConsumer(dest);

ProducerEventSource source = new ProducerEventSource(connection, dest);
System.out.println("Setting Producer Listener");
source.setProducerListener(prodevent -> {
    System.out.println("producer status: " + prodevent.isStarted());
});
// Gets called from inside the broker's Thread and somehow causes deadlocks if I don't invoke this from the outside
new Thread(() -> {
    try {
        consumer.setMessageListener(new NetworkEventPlayerAdapter(objectMapper, event, gameEventManager, playerID));
    } catch (Exception e) {
        e.printStackTrace();
    }
}).start();

不幸的是，当我强行退出之前添加为生产者 (Alt+F4) 的另一个应用程序时，TransportListener 和 ProducerListener 都没有给我任何输出。经纪人肯定注意到了：

 WARN | Transport Connection to: tcp://127.0.0.1:58988 failed: java.net.SocketException: Connection reset
 WARN | Transport Connection to: tcp://127.0.0.1:58986 failed: java.net.SocketException: Connection reset

但我找不到在 Java 中获得这些事件回调的方法。我还尝试在代理中设置自定义 IOExceptionHandler 并向连接添加 ExceptionListener。他们也从未被调用过。

Answer 1

一种可能的方法是为 连接重置

解析 log-output

WARN | Transport Connection to: tcp://127.0.0.1:58988 failed: java.net.SocketException: Connection reset WARN | Transport Connection to: tcp://127.0.0.1:58986 failed: java.net.SocketException: Connection reset

因为套接字是在 activemq 中实现的，所以您必须在那里添加 ExceptionListener 以编写您自己的异常处理例程...

Answer 2

您可以使用咨询主题 ActiveMQ.Advisory.Connection 甚至 ActiveMQ.Advisory.Producer.Queue ActiveMQ.Advisory.Producer.Topic，这些提供了生产者数量连接的统计数据，检查这个 link http://activemq.apache.org/advisory-message.html

当生产者的连接断开时如何得到通知？

How to get notified when a producer's connection dropped?

java

activemq