SpringBoot集成RabbitMQ实现可靠消息投递与延迟任务

@Scheduled(cron = "0/30 * * * * ?")
public void scanAndClose() {
    LocalDateTime now = LocalDateTime.now();
    List<Order> expiredOrders = orderMapper.selectByStatusAndCreateTimeBefore("WAIT_PAY", now.minusMinutes(15));
    for (Order order : expiredOrders) {
        orderService.close(order);
    }
}

@Bean
public Queue waitQueue() {
    return QueueBuilder.durable("order.wait.queue")
            .ttl(15 * 60 * 1000)          // 15 分钟
            .deadLetterExchange("order.dlx.exchange")
            .deadLetterRoutingKey("close")
            .build();
}

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated   # 开启发送确认
    publisher-returns: true              # 开启路由失败回调
    listener:
      simple:
        acknowledge-mode: manual         # 手动 ACK
        retry:
          enabled: true
          max-attempts: 3

@Configuration
public class RabbitConfig {

    // 延迟交换机
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");   // 支持 direct / topic / fanout 等
        return new CustomExchange("order.delay.exchange", "x-delayed-message", true, false, args);
    }

    // 消费者真正监听的队列
    @Bean
    public Queue closeOrderQueue() {
        // 配置死信交换机，方便后续把处理失败的消息转走
        return QueueBuilder.durable("order.close.queue")
                .deadLetterExchange("order.dlx.exchange")
                .deadLetterRoutingKey("close.dead")
                .build();
    }

    // 绑定关系
    @Bean
    public Binding binding(Queue closeOrderQueue, CustomExchange delayExchange) {
        return BindingBuilder.bind(closeOrderQueue)
                .to(delayExchange)
                .with("order.close")
                .noargs();
    }

    // 死信队列，用于接收处理失败的消息，做最终兜底
    @Bean
    public Queue orderCloseDeadQueue() {
        return new Queue("order.close.dlq", true);
    }

    @Bean
    public Binding deadBinding() {
        return BindingBuilder.bind(orderCloseDeadQueue())
                .to(new TopicExchange("order.dlx.exchange"))
                .with("close.dead");
    }
}

@Service
@Slf4j
public class OrderMessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendDelayCloseMessage(Long orderId, long delayMillis) {
        Map<String, Object> message = new HashMap<>();
        message.put("orderId", orderId);
        message.put("type", "ORDER_CLOSE");
        message.put("timestamp", System.currentTimeMillis());

        MessagePostProcessor processor = msg -> {
            // 关键点，设置延迟时间，单位毫秒
            msg.getMessageProperties().setDelay(Long.valueOf(delayMillis).intValue());
            msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return msg;
        };

        rabbitTemplate.convertAndSend(
                "order.delay.exchange",
                "order.close",
                message,
                processor);

        log.info("延迟关闭订单消息已发送, orderId={}, delay={}ms", orderId, delayMillis);
    }
}

@PostConstruct
public void init() {
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            log.error("消息发送失败: {}, cause={}", correlationData.getId(), cause);
            // 可以把消息状态更新到 DB，或者搞个定时任务捞出来重发
        }
    });

    rabbitTemplate.setMandatory(true);
    rabbitTemplate.setReturnsCallback(returned -> {
        log.error("消息路由失败: exchange={}, routingKey={}, message={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getMessage());
    });
}

@Component
@Slf4j
public class OrderCloseConsumer {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private OrderService orderService;

    @RabbitListener(queues = "order.close.queue")
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String content = new String(message.getBody(), StandardCharsets.UTF_8);
        JSONObject json = JSONObject.parseObject(content);
        Long orderId = json.getLong("orderId");

        // 幂等校验，用 Redis 分布式锁或者 SETNX 都行
        String key = "order:close:" + orderId;
        Boolean first = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", Duration.ofHours(1));
        if (!Boolean.TRUE.equals(first)) {
            log.warn("重复消息，直接确认: orderId={}", orderId);
            channel.basicAck(deliveryTag, false);
            return;
        }

        try {
            boolean success = orderService.closeOrder(orderId);
            if (success) {
                channel.basicAck(deliveryTag, false);
                log.info("订单关闭成功: orderId={}", orderId);
            } else {
                // 业务上明确返回失败，比如订单已经被用户手动取消，这种就属于“预期内的失败”，
                // 直接 ack 算了，别再重试了，不然是浪费资源
                channel.basicAck(deliveryTag, false);
                log.warn("订单关闭失败(业务), 不再重试: orderId={}", orderId);
                stringRedisTemplate.delete(key); // 这里要记得删掉幂等标记
            }
        } catch (Exception e) {
            // 真正的异常情况，比如数据库连接挂了，这种可以重试
            long retryCount = message.getMessageProperties().getHeaders()
                    .getOrDefault("retryCount", 0) instanceof Integer ?
                    (Integer) message.getMessageProperties().getHeaders().get("retryCount") : 0;

            if (retryCount < 2) {
                log.warn("订单关闭失败，准备重试, orderId={}, retryCount={}", orderId, retryCount);
                // 这里我用的 requeue=true，让消息重新回到队头
                // 但要注意，重试次数我是放在 header 里的，requeue 回来 header 还会带着
                channel.basicNack(deliveryTag, false, true);
            } else {
                // 超过最大重试次数，拒收消息，让它进死信队列
                log.error("订单关闭失败，进入死信队列: orderId={}", orderId);
                channel.basicNack(deliveryTag, false, false);
            }
        }
    }
}

@Bean
public Queue waitQueue() {
    return QueueBuilder.durable("order.wait.queue")
            .ttl(900000) // 15 分钟
            .deadLetterExchange("order.dlx.exchange")
            .deadLetterRoutingKey("close")
            .build();
}

@Autowired
private MeterRegistry meterRegistry;

// 消费者异常时
meterRegistry.counter("order.close.failures", "queue", "order.close.queue").increment();

// 发送时，把 TraceId 塞进去
message.getMessageProperties().setHeader("traceId", traceIdGenerator.generate());
// 消费时，从消息里取出来放到 MDC
String traceId = message.getMessageProperties().getHeader("traceId");
MDC.put("traceId", traceId);