当前位置：首页 > news >正文

RabbitMQ 断网自动重连失效

news 来源：原创 2025/5/26 6:15:40

问题描述

某天晚上公司告警群里突然有消息堆积的告警，业务人员排查发现业务应用不消费消息了，由于是很多应用都不消费了，初步推断可能 rabbitMQ server 端出了问题，和阿里云技术支持沟通后发现升级了 RabbitMQ server 端，业务应用重启后消费正常了。

问题排查

第二天开始排查具体什么原因导致的不消费，理论上 RabbitMQ client 默认是开启自动重连功能的，为什么没有起作用？通过日志可以看到客户端和RabbitMQ server 端是进行了断网重连：

INFO  o.s.a.r.l.SimpleMessageListenerContainer Restarting Consumer@2590a554: tags=[[]], channel=Cached Rabbit Channel:
INFO  o.s.a.r.c.CachingConnectionFactory Created new connection: rabbitConnectionFactory

既然客户端有重连行为，那么就要继续排查为什么消息停止消费了。后续阿里的专家说可能是因为不同的 consumer 使用了相同的 consumerTag 导致的，之前其他客户也有遇到过。

通过查询 RabbitMQ 官方文档发现，确实当不同的consumer 使用相同的 consumerTag 会导致连接重连会出问题。
在这里插入图片描述
https://www.rabbitmq.com/client-libraries/java-api-guide#consuming

代码核查

先查看公司代码发现当调用 channel.basicConsume() 方法时，不同的 consumer 确实传入的 consumerTag 是一样的，印证了上述的推测。

再继续深究一下，查看 rabbitmq client 的代码，发现调用 channel.basicConsume() 方法时，最终是会把 consumerTag 和 consumer 存储在 AutorecoveringConnection 对象的 Map<String, RecordedConsumer> consumers 里面，consumerTag 是 key，一般情况下 Channel 是共享 Connection 的，所以当 consumerTag 相同时就会存在覆盖的情况。

public String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException {final String result = delegate.basicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, callback);recordConsumer(result, queue, autoAck, exclusive, arguments, callback);return result;
}private void recordConsumer(String result,String queue,boolean autoAck,boolean exclusive,Map<String, Object> arguments,Consumer callback) {RecordedConsumer consumer = new RecordedConsumer(this, queue).autoAck(autoAck).consumerTag(result).exclusive(exclusive).arguments(arguments).consumer(callback);this.consumerTags.add(result);this.connection.recordConsumer(result, consumer);}

 void recordConsumer(String result, RecordedConsumer consumer) {this.consumers.put(result, consumer);}

下面是断网重连的逻辑，会从 consumers map 中取出 consumer 进行消费恢复：

private void recoverTopology(final ExecutorService executor) {// The recovery sequence is the following:// 1. Recover exchanges// 2. Recover queues// 3. Recover bindings// 4. Recover consumersif (executor == null) {// recover entities in serial on the main connection threadfor (final RecordedExchange exchange : Utility.copy(recordedExchanges).values()) {recoverExchange(exchange, true);}for (final Map.Entry<String, RecordedQueue> entry : Utility.copy(recordedQueues).entrySet()) {recoverQueue(entry.getKey(), entry.getValue(), true);}for (final RecordedBinding b : Utility.copy(recordedBindings)) {recoverBinding(b, true);}for (final Map.Entry<String, RecordedConsumer> entry : Utility.copy(consumers).entrySet()) {recoverConsumer(entry.getKey(), entry.getValue(), true);}} else {// Support recovering entities in parallel for connections that have a lot of queues, bindings, & consumers// A channel is single threaded, so group things by channel and recover 1 entity at a time per channel// We also need to recover 1 type of entity at a time in case channel1 has a binding to a queue that is currently owned and being recovered by channel2 for example// Note: invokeAll will block until all callables are completed and all returned futures will be complete try {recoverEntitiesAsynchronously(executor, Utility.copy(recordedExchanges).values());recoverEntitiesAsynchronously(executor, Utility.copy(recordedQueues).values());recoverEntitiesAsynchronously(executor, Utility.copy(recordedBindings));recoverEntitiesAsynchronously(executor, Utility.copy(consumers).values());} catch (final Exception cause) {final String message = "Caught an exception while recovering topology: " + cause.getMessage();final TopologyRecoveryException e = new TopologyRecoveryException(message, cause);getExceptionHandler().handleTopologyRecoveryException(delegate, null, e);}}}