ActiveMQ Artemis 在 HA 故障转移期间发布消息丢失
ActiveMQ Artemis publish message loss during HA fail-over
我使用 ActiveMQ Artemis 2.17.0,我希望在故障转移期间避免生产者中的消息丢失。
通过捕获 ActiveMQUnBlockedException
并再次发送消息,在 Artemis 主动到被动切换期间处理了消息发布丢失。
代理配置为 active/passive HA 共享存储。 host1
中配置的主动节点和 host2
中配置的被动节点。
Url 是:
(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。
按照我为模拟故障转移所做的步骤
- node1 和 node2 已启动
- node1 作为实时服务器启动,node2 作为备份服务器启动
- killed node1, node2 become live server
- 客户端发布代码抛出
ActiveMQUnBlockedException
并处理以再次发送消息
- 再次启动了 node1。 node1 成为实时服务器,node2 再次成为备份
- 客户端发布代码没有抛出
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)
您得到的 ActiveMQUnBlockedException
来自 Spring 对 javax.jms.Connection#stop
的调用。它与发送消息无关。当您收到此特定异常时重新发送消息可能会导致重复消息。
最终您的问题与设置 blockOnDurableSend=false
直接相关。这告诉客户“解雇后忘记”。换句话说,客户端不会等待代理的响应来确保消息确实成功发送。这种等待的减少增加了吞吐量,但降低了可靠性。
如果您真的想减少潜在的消息丢失,您有两个主要选择。
设置blockOnDurableSend=true
。这会降低消息吞吐量,但这是保证消息成功到达代理的最简单方法。
使用 CompletionListener
。这将允许您保留 blockOnDurableSend=false
,但如果发送消息时出现问题,应用程序仍会收到通知,尽管信息将以异步方式提供。 JMS 2 中专门为这种场景添加了此功能。有关详细信息,请参阅 JavaDoc。
我使用 ActiveMQ Artemis 2.17.0,我希望在故障转移期间避免生产者中的消息丢失。
通过捕获 ActiveMQUnBlockedException
并再次发送消息,在 Artemis 主动到被动切换期间处理了消息发布丢失。
代理配置为 active/passive HA 共享存储。 host1
中配置的主动节点和 host2
中配置的被动节点。
Url 是:
(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。 按照我为模拟故障转移所做的步骤
- node1 和 node2 已启动
- node1 作为实时服务器启动,node2 作为备份服务器启动
- killed node1, node2 become live server
- 客户端发布代码抛出
ActiveMQUnBlockedException
并处理以再次发送消息 - 再次启动了 node1。 node1 成为实时服务器,node2 再次成为备份
- 客户端发布代码没有抛出
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)
您得到的 ActiveMQUnBlockedException
来自 Spring 对 javax.jms.Connection#stop
的调用。它与发送消息无关。当您收到此特定异常时重新发送消息可能会导致重复消息。
最终您的问题与设置 blockOnDurableSend=false
直接相关。这告诉客户“解雇后忘记”。换句话说,客户端不会等待代理的响应来确保消息确实成功发送。这种等待的减少增加了吞吐量,但降低了可靠性。
如果您真的想减少潜在的消息丢失,您有两个主要选择。
设置blockOnDurableSend=true
。这会降低消息吞吐量,但这是保证消息成功到达代理的最简单方法。
使用 CompletionListener
。这将允许您保留 blockOnDurableSend=false
,但如果发送消息时出现问题,应用程序仍会收到通知,尽管信息将以异步方式提供。 JMS 2 中专门为这种场景添加了此功能。有关详细信息,请参阅 JavaDoc。