消费者异常恢复

生产环境中消费者可能因网络抖动、Broker重启或自身崩溃而断开连接，自动恢复机制是保障消费链路持续运行的关键。

定义

消费者异常恢复指在连接断开、Channel关闭或消费者进程崩溃后，自动重建连接、重新声明队列与交换机、重新注册消费者的完整流程。

Maven依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.20.0</version>
</dependency>

配置与示例

自动重连消费者

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

public class ResilientConsumer {
    private static final String QUEUE_NAME = "resilient_queue";
    private static final String EXCHANGE_NAME = "resilient_exchange";
    private static final String ROUTING_KEY = "resilient.key";
    
    // 重连参数
    private static final int MAX_RECONNECT_ATTEMPTS = 10;
    private static final long RECONNECT_DELAY_MS = 3000;
    
    private final AtomicBoolean running = new AtomicBoolean(true);
    private volatile Connection connection;
    private volatile Channel channel;

    public void start() {
        int attempts = 0;
        while (running.get() && attempts < MAX_RECONNECT_ATTEMPTS) {
            try {
                connect();
                attempts = 0; // 连接成功，重置计数器
                waitForDisconnect();
            } catch (Exception e) {
                attempts++;
                System.err.println("连接失败 (第" + attempts + "次)，等待重连: " + e.getMessage());
                try {
                    Thread.sleep(RECONNECT_DELAY_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    private void connect() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        
        // 启用自动恢复（RabbitMQ Java Client内置机制）
        factory.setAutomaticRecoveryEnabled(true);
        factory.setTopologyRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);

        connection = factory.newConnection();
        channel = connection.createChannel();

        // 声明拓扑结构（幂等，重复声明无副作用）
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

        // 注册消费者
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("收到消息: " + message);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
        System.out.println("消费者已启动，等待消息...");
    }

    private void waitForDisconnect() throws IOException, TimeoutException {
        // 阻塞等待连接关闭
        connection.addShutdownListener(cause -> {
            System.err.println("连接断开: " + cause.getMessage());
        });
        // 使用CountDownLatch或阻塞调用等待
        while (connection.isOpen()) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    public void stop() {
        running.set(false);
        try {
            if (channel != null && channel.isOpen()) channel.close();
            if (connection != null && connection.isOpen()) connection.close();
        } catch (Exception e) {
            System.err.println("关闭异常: " + e.getMessage());
        }
    }

    public static void main(String[] args) {
        ResilientConsumer consumer = new ResilientConsumer();
        Runtime.getRuntime().addShutdownHook(new Thread(consumer::stop));
        consumer.start();
    }
}

RabbitMQ Java Client内置的AutomaticRecoveryEnabled可自动重建Connection和Channel，但不会重新声明非持久的队列与绑定。必须设置TopologyRecoveryEnabled=true并配合幂等的队列声明逻辑。

注意事项

1. 幂等声明：恢复流程中重复执行queueDeclare和queueBind是安全的，RabbitMQ对已存在的同名队列不会报错。

2. 自动恢复局限：内置自动恢复仅重建Connection/Channel，消费者需重新调用basicConsume注册。

3. 重连间隔控制：使用指数退避策略避免频繁重连。例如首次等待3秒，下次6秒、12秒，上限60秒。

4. 未确认消息处理：连接断开时未ACK的消息会被Broker重新入队，恢复后可能被其他消费者投递，需保证消费逻辑幂等。

5. 关闭钩子：通过Runtime.getRuntime().addShutdownHook确保进程退出时优雅关闭连接，避免消息处于半处理状态。

要点总结

• 连接断开后需自动重建Connection、Channel并重新注册消费者
• 启用setAutomaticRecoveryEnabled(true)和setTopologyRecoveryEnabled(true)
• 队列与交换机声明必须幂等，重复执行无副作用
• 重连应采用指数退避策略，避免频繁重试冲击Broker
• 消费逻辑必须幂等，应对连接断开导致的消息重复投递