Spring Boot如何限制RabbitMQ、PostgreSQL等连接失败后的重试次数

针对@RabbitListener连接失败无限重试的问题，有两种成熟的实现方案，可根据业务灵活选择：

方案1：基于application.yml原生配置（快速实现） #

Spring AMQP 2.3及以上版本原生支持连接工厂层面的重试参数配置，无需额外写代码即可实现重试次数限制：

spring:
  rabbitmq:
    addresses: 你的RabbitMQ地址:5672
    username: 账号
    password: 密码
    # 连接阶段重试配置
    connection-factory:
      retry:
        enabled: true
        max-attempts: 3 # 最大重试次数
        initial-interval: 1000ms # 首次重试间隔
        multiplier: 2 # 间隔倍增系数
        max-interval: 5000ms # 最大重试间隔
    listener:
      simple:
        missing-queues-fatal: true # 达到重试上限后停止监听器初始化，不再无限重试
        # 以下是业务消息消费的重试配置，和连接重试无关，不需要可关闭
        retry:
          enabled: false

该方案适合不需要自动恢复监听的场景，达到重试上限后会停止监听器启动流程，避免持续打印错误日志。

方案2：自定义监听器实现可自动恢复的重试控制 #

如果需要3次重试失败后暂时停止监听，后续RabbitMQ服务恢复后自动重启监听，可以通过自定义事件处理器+定时检测实现：

步骤1：自定义连接失败事件处理器，统计重试次数并停止监听器 #

import org.springframework.amqp.rabbit.connection.ConnectionFailedEvent;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class RabbitConnectionFailureHandler implements ApplicationListener<ConnectionFailedEvent> {
    private final AtomicInteger retryCounter = new AtomicInteger(0);
    private static final int MAX_RETRY_COUNT = 3;
    @Resource
    private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

    @Override
    public void onApplicationEvent(ConnectionFailedEvent event) {
        int currentCount = retryCounter.incrementAndGet();
        if (currentCount >= MAX_RETRY_COUNT) {
            // 达到重试上限，停止所有运行中的Rabbit监听器
            rabbitListenerEndpointRegistry.getListenerContainers().forEach(container -> {
                if (container.isRunning()) {
                    container.stop();
                }
            });
            retryCounter.set(0);
            // 可在此处添加自定义告警逻辑，通知运维人员RabbitMQ连接异常
        }
    }
}

步骤2：新增定时任务检测连接状态，服务恢复后自动重启监听器 #

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;

@Component
public class RabbitConnectionRecoveryTask {
    @Resource
    private ConnectionFactory connectionFactory;
    @Resource
    private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

    // 每30秒检测一次连接状态，可根据业务调整间隔
    @Scheduled(fixedDelay = 30000)
    public void checkAndRecoverListener() {
        // 没有停止的监听器直接跳过
        boolean hasStoppedContainer = rabbitListenerEndpointRegistry.getListenerContainers()
                .stream().anyMatch(container -> !container.isRunning());
        if (!hasStoppedContainer) {
            return;
        }
        // 尝试连接判断RabbitMQ是否恢复
        try {
            connectionFactory.createConnection().close();
            // 连接正常，重启所有监听器
            rabbitListenerEndpointRegistry.getListenerContainers().forEach(container -> {
                if (!container.isRunning()) {
                    container.start();
                }
            });
        } catch (Exception e) {
            // 连接仍异常，等待下次检测即可
        }
    }
}

使用该方案需要在启动类上添加@EnableScheduling注解开启定时任务能力。

注意事项 #

• 注意区分连接重试和业务消息消费重试：上述配置均针对RabbitMQ服务连接建立阶段的重试，业务逻辑处理消息抛出异常的重试由spring.rabbitmq.listener.simple.retry配置单独控制，二者不要混淆
• 多实例部署时建议给定时任务的执行时间加随机偏移，避免所有实例同时发起连接请求压垮刚恢复的RabbitMQ服务

内容的提问来源于stack exchange，提问作者Alyan Ahmed