Vert.x AMQP客户端在Broker宕机时无法自动重连问题排查

你的代码无法触发Broker宕机重连，是因为遗漏了3项必要配置和逻辑：

• 未开启客户端自动重连能力：Vert.x AMQP客户端的自动重连默认是关闭状态，你没有配置重连次数、重连间隔相关参数，连接断开后客户端不会主动发起重试。
• 异步API调用逻辑错误：amqpClient.connect()是异步非阻塞方法，你在发起连接请求后没有等待连接建立完成，就直接调用createReceiver方法，不仅接收器创建逻辑不可靠，连接状态变化时也无法关联到已创建的消费链路。
• 缺少连接/链路的状态监听：没有给连接、接收器绑定异常、关闭事件的处理回调，Broker宕机触发连接断开时，没有任何逻辑会触发你的错误处理分支，程序自然不会打印日志或执行重连动作。

第一步：补充重连相关配置 #

在AmqpClientOptions中添加重连参数，开启自动重连能力：

AmqpClientOptions options = new AmqpClientOptions()
        .setHost("localhost")
        .setPort(5672)
        .setUsername("")
        .setPassword("")
        // 配置最大重连次数，设为Integer.MAX_VALUE表示无限次重连，可按需调整为固定值
        .setReconnectAttempts(Integer.MAX_VALUE)
        // 两次重连的间隔时间，单位毫秒，这里设为1秒
        .setReconnectInterval(1000)
        // 单次连接的超时时间，单位毫秒
        .setConnectTimeout(5000);

第二步：调整异步调用顺序，添加状态监听 #

必须在连接建立成功的回调内，基于拿到的连接实例创建消息接收器，同时给连接、接收器绑定事件监听，感知状态变化。完整修正后的代码如下：

public class BrokerConnector {

    public void consumeEventsQueue() {
        AmqpClientOptions options = new AmqpClientOptions()
                .setHost("localhost")
                .setPort(5672)
                .setUsername("")
                .setPassword("")
                .setReconnectAttempts(Integer.MAX_VALUE)
                .setReconnectInterval(1000)
                .setConnectTimeout(5000);

        AmqpClient amqpClient = AmqpClient.create(options);
        
        // 封装连接+订阅逻辑，首次启动和异常场景可复用
        Runnable startConnect = () -> amqpClient.connect(conRes -> {
            if (conRes.failed()) {
                System.out.println("连接Broker失败，客户端将按配置自动重试，异常原因：" + conRes.cause().getMessage());
                return;
            }
            System.out.println("Broker连接建立成功");
            AmqpConnection connection = conRes.result();

            // 监听连接关闭事件
            connection.closeHandler(v -> {
                System.out.println("Broker连接已断开，客户端即将发起重连");
            });
            // 监听连接异常事件
            connection.exceptionHandler(err -> {
                System.out.println("Broker连接发生异常：" + err.getMessage());
            });

            // 连接成功后再创建队列接收器
            connection.createReceiver("MY_QUEUE", receiverRes -> {
                if (receiverRes.failed()) {
                    System.out.println("创建消息接收器失败，将触发重连，异常原因：" + receiverRes.cause().getMessage());
                    // 主动关闭当前异常连接，触发客户端重连逻辑
                    connection.close();
                    return;
                }
                AmqpReceiver receiver = receiverRes.result();
                System.out.println("MY_QUEUE队列接收器创建成功，开始消费消息");

                // 监听接收器异常与关闭事件
                receiver.exceptionHandler(err -> {
                    System.out.println("消息接收器发生异常：" + err.getMessage());
                });
                receiver.closeHandler(v -> {
                    System.out.println("消息接收器已关闭");
                });

                // 消息消费逻辑
                receiver.handler(msg -> {
                    System.out.println("收到消息：" + msg.bodyAsString());
                    // 消费成功后手动确认消息，避免消息重复投递
                    msg.accept();
                });
            });
        });

        startConnect.run();
    }
}

• 不要直接调用amqpClient.createReceiver()的简化方法创建接收器：该方法会隐式创建内部连接，你无法拿到连接实例做状态监听，重连过程的状态完全不可控，生产环境建议显式调用connect拿到连接实例后再创建生产者、消费者。
• 如果使用集群部署的Broker，可通过AmqpClientOptions添加多个Broker节点地址，重连时客户端会自动尝试连接可用节点，实现故障切换。
• 开启自动重连后，客户端在重连成功后会自动恢复之前在连接上创建的消费链路，不需要你手动重复调用接收器创建逻辑。

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