消费端可靠性

消费端处理消息时可能遇到异常、超时或服务重启等情况。合理的异常处理与重试策略可避免消息丢失或无限重试导致的服务雪崩。

消息确认模式

自动确认（autoAck = true）

channel.basicConsume(QUEUE_NAME, true, (consumerTag, delivery) -> {
    // 消息一旦投递即视为已消费
    // 若处理失败，消息已丢失
}, consumerTag -> {});

注意：自动确认性能高但不可靠，消息在消费者处理失败时无法恢复，仅适用于允许丢失的场景。

手动确认（autoAck = false）

// 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.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class ManualAckExample {
    
    private static final String QUEUE_NAME = "manual_ack_queue";
    
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            
            // autoAck = false 手动确认
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                try {
                    String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                    System.out.println("收到消息: " + message);
                    
                    // 业务处理
                    processMessage(message);
                    
                    // 处理成功，确认消息
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    System.out.println("消息已确认: " + delivery.getEnvelope().getDeliveryTag());
                    
                } catch (Exception e) {
                    System.err.println("消息处理失败: " + e.getMessage());
                    // 拒绝消息，重新入队
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                }
            };
            
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
        }
    }
    
    private static void processMessage(String message) {
        // 模拟业务处理
        if (message.contains("error")) {
            throw new RuntimeException("处理异常");
        }
    }
}

消息拒绝与重试

basicReject 与 basicNack

重试策略

public class RetryStrategy {
    
    private static final int MAX_RETRY = 3;
    
    public static void consumeWithRetry(Channel channel) 
            throws IOException, InterruptedException {
        
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            long deliveryTag = delivery.getEnvelope().getDeliveryTag();
            
            try {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                processMessage(message);
                
                // 成功：确认消息
                channel.basicAck(deliveryTag, false);
                
            } catch (Exception e) {
                int retryCount = getRetryCount(delivery);
                
                if (retryCount < MAX_RETRY) {
                    // 未达到最大重试次数，重新入队
                    channel.basicNack(deliveryTag, false, true);
                    System.out.println("消息重新入队，重试次数: " + (retryCount + 1));
                } else {
                    // 超过最大重试次数，拒绝消息不重新入队
                    channel.basicNack(deliveryTag, false, false);
                    System.err.println("消息超过最大重试次数，已丢弃或转死信");
                    // 可在此处记录日志或转发到死信队列
                }
            }
        };
        
        channel.basicConsume("retry_queue", false, deliverCallback, consumerTag -> {});
    }
    
    private static int getRetryCount(Delivery delivery) {
        // 从消息头中获取重试次数
        Object count = delivery.getProperties().getHeaders().get("x-retry-count");
        return count != null ? (Integer) count : 0;
    }
    
    private static void processMessage(String message) throws Exception {
        // 模拟可能失败的业务逻辑
        throw new RuntimeException("业务异常");
    }
}

失败消息降级处理

降级策略设计

转存死信队列示例

public class DeadLetterFallback {
    
    private static final String DLX_EXCHANGE = "dlx_exchange";
    private static final String DLX_QUEUE = "dlx_queue";
    
    public static void setupDeadLetterQueue(Channel channel) 
            throws IOException {
        // 声明死信交换机
        channel.exchangeDeclare(DLX_EXCHANGE, "direct", true);
        // 声明死信队列
        channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, "dlx.route");
    }
    
    public static void sendToDeadLetter(Channel channel, Delivery delivery) 
            throws IOException {
        long deliveryTag = delivery.getEnvelope().getDeliveryTag();
        
        // 拒绝消息，不重新入队（requeue=false）
        // 若队列配置了 x-dead-letter-exchange，消息自动转存 DLX
        channel.basicNack(deliveryTag, false, false);
        
        System.err.println("消息已转存死信队列");
    }
}

带重试次数的消息头传递

public class RetryWithHeader {
    
    public static void republishWithRetryCount(Channel channel, 
            Delivery delivery, String exchange, String routingKey) 
            throws IOException {
        
        // 获取当前重试次数
        int retryCount = getRetryCount(delivery);
        retryCount++;
        
        // 构建新消息属性，携带重试次数
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(java.util.Map.of("x-retry-count", retryCount))
                .build();
        
        // 重新发布消息
        channel.basicPublish(exchange, routingKey, props, delivery.getBody());
        
        // 确认原消息
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

注意事项

1. requeue = true 时消息会重新入队到原队列头部，可能无限重试同一消息导致队列阻塞。
2. 必须设置最大重试次数，避免消息因持续失败而无限重试，消耗系统资源。
3. 消费端异常应区分类型：网络异常可重试，业务异常（如参数校验失败）重试无意义。
4. 手动确认模式下，若消费者未确认即断开连接，消息会自动重新入队。
5. 消息确认应在业务逻辑成功后执行，避免先确认后处理导致消息丢失。

要点总结

• 手动确认（autoAck=false）是消费端可靠性的基础，避免消息处理失败后丢失
• basicNack(requeue=true) 实现消息重试，需配合最大重试次数限制
• 超过重试次数的消息应降级处理：丢弃、记录日志或转存死信队列
• 重试次数可通过消息头 x-retry-count 传递，避免无限重试
• 消费端异常需分类处理，业务异常重试无意义时应直接拒绝