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

如何解决ActiveMQ Artemis 在 HA 故障转移期间发布消息丢失

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

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

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

blockOnDurableSend 设置为 false 以实现高吞吐量。 在主动到被动切换期间，发布代码抛出 ActiveMQUnBlockedException，但在被动到主动切换期间不会。

我们使用 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成为live server
4. 客户端发布代码抛出 ActiveMQUnBlockedException 并处理以再次发送消息
5. 再次启动 node1。 node1 成为 live server，node2 再次成为备份
6. 客户端发布代码未抛出 ActiveMQUnBlockedException 并丢失消息。

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

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)

解决方法

您获得的 ActiveMQUnBlockedException 来自 Spring 对 javax.jms.Connection#stop 的调用。它与发送消息无关。当您收到此特定异常时重新发送消息可能会导致重复消息。

最终，您的问题与设置 blockOnDurableSend=false 直接相关。这告诉客户端“解雇并忘记”。换句话说，客户端不会等待来自代理的响应来确保消息确实成功发送。无需等待会增加吞吐量，但会降低可靠性。

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

设置 blockOnDurableSend=true。这会降低消息吞吐量，但这是保证消息成功到达代理的最简单方法。

使用 CompletionListener。这将允许您保留 blockOnDurableSend=false，但如果发送消息出现问题，应用程序仍将收到通知，尽管信息将异步提供。 JMS 2 中专门为这种场景添加了此功能。有关更多详细信息，请参阅 JavaDoc。