spring:
  rabbitmq:
    host: 192.168.1.100
    port: 5672
    virtual-host: /
    username: admin
    password: 123456
    
    # ===== 重连配置 =====
    connection-timeout: 1s          # 建立连接超时时间
    template:
      retry:
        enabled: true               # 开启 RabbitTemplate 的重试
        initial-interval: 1000ms    # 首次重试等待 1s
        multiplier: 2               # 每次重试等待时间 * 2（指数退避） 下次等待=initial*mult
        max-attempts: 3             # 最多重试 3 次（含首次）
        max-interval: 10000ms       # 重试等待上限 10s

/**
 * 连接层面的自动恢复配置
 * Spring AMQP 底层 AMQP Client 默认已开启 automatic recovery
 * 但生产环境建议显式配置，更可控
 */
@Configuration
public class RabbitConnectionConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("192.168.1.100");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("/");

        // 连接缓存模式：CHANNEL（默认）vs CONNECTION
        // CHANNEL 模式：一个物理连接，多个 channel 复用
        // CONNECTION 模式：每次 getConnection() 可能返回新连接（适合高并发）
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        factory.setChannelCacheSize(10); // channel 缓存数量

        return factory;
    }
}

spring:
  rabbitmq:
    publisher-confirm-type: correlated  # 开启 CORRELATED 异步确认
    publisher-returns: true             # 开启 Return 退回回调
    template:
      mandatory: true                   # 消息路由失败时退回而非丢弃（触发 Return）

/**
 * Publisher Confirms + Returns 完整实现
 */
@Slf4j
@Configuration  
public class PublisherConfirmConfig implements RabbitTemplate.ConfirmCallback,
                                               RabbitTemplate.ReturnsCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 注意：RabbitTemplate 是 prototype 的，但这里通过 @PostConstruct
     * 把 callback 注入进去，每个 RabbitTemplate 实例只设置一次
     */
    @PostConstruct
    public void init() {  //接口的子实现类, 所以init可以直接this
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * Confirm 回调
     * 消息成功到达 Exchange → ack=true
     * Broker 内部错误     → ack=false, cause 说明原因
     *
     * @param correlationData 发送时传入的关联数据（含消息ID）
     * @param ack             是否确认成功
     * @param cause           失败原因（ack=true 时为 null）
     */
     
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String msgId = correlationData != null ? correlationData.getId() : "unknown";

        if (ack) {
            log.info("[Confirm ACK] 消息已到达Exchange, msgId={}", msgId);
            // 可选：更新数据库消息状态为"已投递"
        } else {
            log.error("[Confirm NACK] 消息投递Exchange失败, msgId={}, cause={}", msgId, cause);
            // 触发补偿：重发消息 / 告警
            // 注意：这里是异步线程，不能抛异常影响主流程
            handleSendFailed(msgId, cause);
        }
    }

    /**
     * Return 回调（路由失败时触发）
     * 消息到了 Exchange 但没有匹配的 Queue binding
     *
     * @param returned 退回的消息详情（含原始 Message、routingKey、replyText 等）
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("[Return] 消息路由失败被退回! exchange={}, routingKey={}, replyCode={}, replyText={}, body={}",
                returned.getExchange(),
                returned.getRoutingKey(),
                returned.getReplyCode(),
                returned.getReplyText(),
                new String(returned.getMessage().getBody()));
        // 处理：检查 routingKey / binding 配置是否正确
        // 生产建议：写入 DB 做人工复查，或者投递到专用的 unroutable 队列
    }

    private void handleSendFailed(String msgId, String cause) {
        // 示例：从 DB 捞出消息体重新发送
        // messageService.markFailed(msgId);
        // alertService.send("MQ投递失败: " + msgId);
    }
}

/**
 * 持久化相关配置代码
 */
@Configuration
public class PersistenceConfig {

    
    @Bean  // Exchange 持久化 
    public DirectExchange orderExchange() {
        // 三个参数：name, durable, autoDelete
        return new DirectExchange("order.direct", true, false);
    }

    
    @Bean  // Queue 持久化
    public Queue orderQueue() {
        return QueueBuilder
                .durable("order.queue")  // durable=true
                // .nonDurable()         // durable=false（重启后消失，不要用在生产）
                .build();
    }

    /**
     * 全局消息转换器配置
     * Jackson2JsonMessageConverter 序列化时默认设置 deliveryMode=PERSISTENT
     * 这是 Spring AMQP 2.x 之后的默认行为
     */
    @Bean
    public MessageConverter messageConverter() {
        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
        // 创建时配置 ObjectMapper，处理日期/空值等
        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
        typeMapper.setTrustedPackages("*"); // 反序列化时信任所有包
        converter.setJavaTypeMapper(typeMapper);
        return converter;
    }

    /**
     * 配置 RabbitTemplate 使用上面的转换器
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                          MessageConverter messageConverter) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(messageConverter);
        return template;
    }
}

/**
 * LazyQueue 声明方式
 * RabbitMQ 3.6+ 支持 x-queue-mode=lazy
 * RabbitMQ 3.12+ LazyQueue 成为默认模式
 */
@Configuration
public class LazyQueueConfig {

    /**
     * 方式一：通过 QueueBuilder（推荐）
     */
    @Bean
    public Queue lazyOrderQueue() {
        return QueueBuilder
                .durable("order.lazy.queue")
                .lazy()  // 等价于 withArgument("x-queue-mode", "lazy")
                .build();
    }

    /**
     * 方式二：手动 withArgument（老版本 Spring AMQP 没有 .lazy() 方法时用）
     */
    @Bean
    public Queue lazyQueueManual() {
        return QueueBuilder
                .durable("order.lazy.queue.v2")
                .withArgument("x-queue-mode", "lazy")
                .build();
    }

    /**
     * 方式三：LazyQueue + 死信队列一起配
     * 生产上推荐这种完整配置
     */
    @Bean
    public Queue lazyQueueWithDLX() {
        return QueueBuilder
                .durable("order.main.queue")
                .lazy()
                .withArgument("x-dead-letter-exchange", "order.dlx.exchange")
                .withArgument("x-dead-letter-routing-key", "order.dlx")
                .withArgument("x-message-ttl", 300000)  // 5 分钟 TTL
                .build();
    }
}

spring:
	rabbitmq:
		listener:
			simple:
				prefetch: 1
				acknowledge-mode : none #none, 关闭ack; manual, 手动ack; auto, 自动ack

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto   # 重试机制下一般配 auto，让 Spring 接管
        retry:
          enabled: true                  # 开启消费者本地重试
          initial-interval: 1000ms       # 首次重试等待 1s
          multiplier: 2                  # 指数退避：1s → 2s → 4s
          max-attempts: 3                # 最多重试 3 次（含首次消费，实际重试 2 次）
          max-interval: 10000ms          # 重试等待上限 10s
          stateless: true                # 无状态重试（不依赖事务），一般用这个

@Configuration
public class RetryConfig {

    /**
     * 推荐方案：RepublishMessageRecoverer
     * 重试耗尽后，把消息投递到一个专门的"失败队列"
     * 保留消息 + 异常信息 + 原始队列信息，方便人工排查
     */
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
        // 参数1：失败后投递到的 exchange
        // 参数2：routing key
        return new RepublishMessageRecoverer(
                rabbitTemplate,
                "error.exchange",   // 专门收失败消息的 exchange
                "error.routing.key"
        );
    }

    /**
     * 配套声明：error exchange 和 error queue
     */
    @Bean
    public DirectExchange errorExchange() {
        return new DirectExchange("error.exchange");
    }

    @Bean
    public Queue errorQueue() {
        return QueueBuilder.durable("error.queue").build();
    }

    @Bean
    public Binding errorBinding() {
        return BindingBuilder.bind(errorQueue())
                .to(errorExchange())
                .with("error.routing.key");
    }
}

/**
 * 配合重试机制的消费者写法（auto ack 模式）
 * Spring 会自动重试，无需手动 channel.basicNack
 */
@Slf4j
@Component
public class OrderConsumerWithRetry {

    @RabbitListener(queues = "order.queue")
    public void consume(OrderDTO order, Message message) {
        log.info("[消费] orderId={}", order.getOrderId());

        // 直接抛异常即可，Spring 的 RetryInterceptor 会捕获并重试
        // 重试次数耗尽后，MessageRecoverer 处理（进 error.queue）
        orderService.processOrder(order);

        // 正常返回 → Spring 自动 ack
        log.info("[完成] orderId={}", order.getOrderId());
    }
}

//注解创建
@RabbitListener( 
	bindings = @QueueBinding(
		value=@Queue(name=, durable=),
					exchange=@Exchange(name=,delayed=true),
					key=""
		)
	)
	
//bean创建
@Bean
public dirEx delayExchange(){
	new ExchangeBuilder
	.directExchange()
	.delayed()
	.durable(true)
	.build();
}

//2.发送消息，利用消息后置处理器添加消息头
@Test
void testPublisherDelayMessage() {
//1.创建消息
String message = "hello, delayed message";
rabbitTemplate.convertAndSend("delay.direct","delay",message,new MessagePostProcessor(){
	@Override
	public Message postProcessMessage(Message message) throws AmqpException {
	//添加延迟消息属性
	message·getMessageProperties().setDelay(5000);
	return message;
	});
}