HA故障转移期间ActiveMQ Artemis发布消息丢失

Question

我使用ActiveMQ Artemis 2.17.0，我希望避免在故障转移期间生成器中丢失消息。

通过捕获ActiveMQunblockedException并再次发送消息来处理Artemis主动到被动切换期间的消息发布丢失。代理被配置为主动/被动HA共享存储。主动节点配置在host1和被动节点配置在host2中。Url为：

(tcp://host1:61616,tcp://host2:61616)?ha=true&reconnectAttempts=-1&blockOnDurableSend=false

我们使用Spring 4.2.5和CachingConnectionFactory作为连接工厂。

我使用以下代码发送消息：

private void sendMessageInternal(final ConnectionFactory connectionFactory, final Destination queue, final String message)
        throws JMSException {

    try (final Connection connection = connectionFactory.createConnection();) {
        connection.start();
        try (final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                final MessageProducer producer = session.createProducer(queue);) {
            final TextMessage textMessage = session.createTextMessage(message);
            producer.send(textMessage);
        }

    } catch (JMSException thr) {
        if (thr.getCause() instanceof ActiveMQUnBlockedException) {
            // consider as fail-over disconnection, send message again.
        } else {
            throw thr;
        }
    }
}

在host1机器中，Artemis部署为master-node1。在host2机器中，Artemis作为slave-node2部署。下面是模拟故障转移的步骤

1. Node1和node2已启动
2. Node1作为实时服务器启动，node2作为备份服务器启动
3. 杀死node1，node2成为实时服务器
4. 客户端发布代码抛出ActiveMQunblockedException并处理以再次发送消息
5. 再次启动node1。node1成为实时服务器，node2再次成为备份
6. 客户端发布代码未引发ActiveMQunblockedException和消息丢失。

在第3步中获得以下错误堆栈。（被杀死的node1和node2成为实时服务器）。

javax.jms.JMSException: AMQ219016: Connection failure detected. Unblocking a blocking call that will never get a response
    at org.apache.activemq.artemis.core.protocol.core.impl.ChannelImpl.sendBlocking(ChannelImpl.java:540)
    at org.apache.activemq.artemis.core.protocol.core.impl.ChannelImpl.sendBlocking(ChannelImpl.java:434)
    at org.apache.activemq.artemis.core.protocol.core.impl.ActiveMQSessionContext.sessionStop(ActiveMQSessionContext.java:470)
    at org.apache.activemq.artemis.core.client.impl.ClientSessionImpl.stop(ClientSessionImpl.java:1121)
    at org.apache.activemq.artemis.core.client.impl.ClientSessionImpl.stop(ClientSessionImpl.java:1110)
    at org.apache.activemq.artemis.jms.client.ActiveMQSession.stop(ActiveMQSession.java:1244)
    at org.apache.activemq.artemis.jms.client.ActiveMQConnection.stop(ActiveMQConnection.java:339)
    at org.springframework.jms.connection.SingleConnectionFactory$SharedConnectionInvocationHandler.localStop(SingleConnectionFactory.java:644)
    at org.springframework.jms.connection.SingleConnectionFactory$SharedConnectionInvocationHandler.invoke(SingleConnectionFactory.java:577)
    at com.sun.proxy.$Proxy5.close(Unknown Source)
    at com.eu.amq.failover.test.ProducerNodeTest.sendMessageInternal(ProducerNodeTest.java:133)
    at com.eu.amq.failover.test.ProducerNodeTest.sendMessage(ProducerNodeTest.java:110)
    at com.eu.amq.failover.test.ProducerNodeTest.main(ProducerNodeTest.java:90)

Answer 1

您得到的ActiveMQunblockedException来自Spring对javax.jms.Connection#Stop的调用。和发送信息无关。当收到此特定异常时，重新发送消息可能会导致重复的消息。

最终，您的问题与设置BlockonDurableSend=false直接相关。这告诉客户“开火并忘记”。换句话说，客户机不会等待代理的响应来确保消息真正成功。这种缺乏等待的情况增加了吞吐量，但降低了可靠性。

如果您真的想减少潜在的消息丢失，您有两个主要的选择。

使用CompletionListener。这将允许您保留BlockonDurableSend=false，但是如果发送消息时出现问题，应用程序仍然会得到通知，尽管信息将异步提供。这个特性是在JMS2中专门为这类场景添加的。有关更多细节，请参见JavaDoc。