详解RabbitMQ队列持久化失败的问题以及如何保障消息可靠性的方法

// Spring Boot配置示例（错误示范）
@Bean
public Queue myQueue() {
    // 缺失durable参数将导致队列非持久化
    return new Queue("orderQueue");
}

2023-06-15 14:23:18 [WARN] Channel shutdown: 
    connection error; protocol method: #method<connection.close>
    (reply-code=406, reply-text=PRECONDITION_FAILED - 
    cannot redeclare exchange 'order_exchange' in vhost '/' with 
    different type, durable parameters)

// Spring Boot正确配置示例
@Configuration
public class RabbitConfig {
    
    // 声明持久化交换机
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange", true, false);
    }

    // 声明持久化队列
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true);
    }

    // 绑定关系配置
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
               .to(orderExchange()).with("order.routingKey");
    }
}

// 生产者消息确认配置
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMandatory(true);
    
    // 确认消息到达Broker
    template.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            // 记录发送失败的消息ID
            log.error("消息未到达交换机: {}", correlationData.getId());
        }
    });
    
    // 确认消息路由到队列
    template.setReturnsCallback(returned -> {
        log.error("消息无法路由到队列: {}", returned.getMessage().getBody());
    });
    
    return template;
}

// 消费者手动确认示例
@RabbitListener(queues = "order.queue")
public void handleOrder(Order order, Channel channel, 
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 业务处理逻辑
        processOrder(order);
        
        // 成功处理，确认消息
        channel.basicAck(tag, false);
    } catch (Exception e) {
        // 处理失败，拒绝消息（不重新入队）
        channel.basicReject(tag, false);
        // 记录异常消息到补偿表
        log.error("订单处理失败: {}", order.getId());
    }
}

// 备份交换器配置
@Bean
public DirectExchange mainExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("alternate-exchange", "ae.exchange");
    return new DirectExchange("main.exchange", true, false, args);
}

@Bean
public FanoutExchange aeExchange() {
    return new FanoutExchange("ae.exchange", true, false);
}

@Bean
public Queue aeQueue() {
    return new Queue("ae.queue", true);
}

# 设置镜像队列策略
rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'