RabbitMQ：连接恢复机制

Question

我正在使用 rabbit mq 3.4.1 java 客户端库，但无法使自动恢复机制工作。

这就是我创建rabbit mq连接工厂的方式：

factory = new ConnectionFactory();
factory.setUsername(userName);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5);
factory.setRequestedHeartbeat(3);

发布消息后，如果我关闭rabbit mq 代理并再次启动它，我希望恢复机制启动并将连接恢复到“正常”状态。但我收到以下错误：

com.rabbitmq.client.AlreadyClosedException: connection is already closed due to connection error; protocol method: #method<connection.close>(reply-code=320, reply-text=CONNECTION_FORCED - broker forced connection closure with reason 'shutdown', class-id=0, method-id=0)
    at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:190) ~[amqp-client-3.4.1.jar:na]
    at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:291) ~[amqp-client-3.4.1.jar:na]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:654) ~[amqp-client-3.4.1.jar:na]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:631) ~[amqp-client-3.4.1.jar:na]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:622) ~[amqp-client-3.4.1.jar:na]

我在这里遗漏了什么吗？解决此问题的唯一方法是注册一个 ShutDownListener 并重新初始化 rabbit mq 连接工厂、连接和通道。

还要回答

“克里斯洛特”

评论，我看到自动恢复开始恢复。我使用临时渠道创建了一个交换：

Channel channel = connection.createChannel();
channel.exchangeDeclare(exchangeName, exchangeType, durable);
channel.close();

我在尝试恢复拓扑时看到以下异常：

Caught an exception when recovering topology Caught an exception while recovering exchange testSuccessfulInitVirtualHost_Exchange: channel is already closed due to clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
com.rabbitmq.client.TopologyRecoveryException: Caught an exception while recovering exchange testSuccessfulInitVirtualHost_Exchange: channel is already closed due to clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
    at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverExchanges(AutorecoveringConnection.java:482)
    at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverEntities(AutorecoveringConnection.java:467)
    at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.beginAutomaticRecovery(AutorecoveringConnection.java:411)
    at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.access$000(AutorecoveringConnection.java:52)
    at com.rabbitmq.client.impl.recovery.AutorecoveringConnection$1.shutdownCompleted(AutorecoveringConnection.java:351)
    at com.rabbitmq.client.impl.ShutdownNotifierComponent.notifyListeners(ShutdownNotifierComponent.java:75)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:574)

如果我不关闭用于创建交换的通道，则不会出现上述异常。

Answer 1

我对 RabbitMQ ConnectionFactory#setAutomaticRecoveryEnabled(Boolean) 方法的解读是，它主要支持从 NETWORK 故障中恢复。

这是一个很好的讨论：https://www.rabbitmq.com/api-guide.html

例如，如果您的机器在一段时间内丢失了到代理的路由，可能是由于交换机或其他故障，那么自动恢复可以重新建立连接等。文档没有说明在经纪人关闭/重启后幸存下来，我认为您的期望不合理。

恕我直言，要从代理重启中恢复，shutdown-listener 方法似乎是一种可靠的方法。

Answer 2

通常，rabbit 客户端应该自己处理恢复 - 您不应该手动重新实现。至少尝试使用lyra。

我在故障转移测试期间遇到了一些问题。连接往往会在代理重新启动时永远挂起，因此关闭信号异常是日志中的最后一件事。我通过设置修复它：

factory.setConnectionTimeout(20000);

临时队列也不能很好地恢复。如果你有这些，你可能需要做一些额外的处理（同样，首先尝试 lyra）。