SpringBoot实现延迟消息的两种方案及对象详解

<!-- RabbitMQ AMQP 核心依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    # 基础连接配置
    host:127.0.0.1
    port:5672
    username:guest
    password:guest
    virtual-host:/
    # 开启生产者确认
    publisher-confirm-type:correlated
    publisher-returns:true
    # 消费者手动ACK
    listener:
      simple:
        acknowledge-mode:manual
        retry:
          enabled:true
          max-attempts: 3

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
 * 延迟消息队列配置类
 * 基于 rabbitmq-delayed-message-exchange 插件实现
 */
@Configuration
publicclassDelayRabbitConfig {
    // 延迟交换机名称
    publicstaticfinalStringDELAY_EXCHANGE="business_delay_exchange";
    // 延迟队列名称
    publicstaticfinalStringDELAY_QUEUE="business_delay_queue";
    // 路由键
    publicstaticfinalStringDELAY_ROUTING_KEY="business.delay.routing";
    /**
     * 构建延迟交换机
     * x-delayed-message：延迟交换机类型
     * x-delayed-type：转发模式（direct/topic/fanout）
     */
    @Bean
    public DirectExchange delayExchange() {
        Map<String, Object> args = newHashMap<>();
        // 核心参数：声明为延迟交换机
        args.put("x-delayed-type", "direct");
        // 参数：名称、持久化、不自动删除、自定义参数
        returnnewDirectExchange(DELAY_EXCHANGE, true, false, args);
    }
    /**
     * 延迟队列（持久化）
     */
    @Bean
    public Queue delayQueue() {
        returnnewQueue(DELAY_QUEUE, true);
    }
    /**
     * 队列与延迟交换机绑定
     */
    @Bean
    public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {
        return BindingBuilder.bind(delayQueue)
                .to(delayExchange)
                .with(DELAY_ROUTING_KEY);
    }
    /**
     * 自定义RabbitTemplate，开启消息可靠投递
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplaterabbitTemplate=newRabbitTemplate(connectionFactory);
        // 开启 mandatory，消息投递失败返回回调
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }
}

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
publicclassDelayMsgProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 发送延迟消息接口
     * @param message 消息内容
     * @param delayTime 延迟时间（单位：毫秒）
     * @return 结果
     */
    @GetMapping("/send/delay/message")
    public String sendDelayMessage(@RequestParam String message,
                                   @RequestParam Long delayTime) {
        // 发送延迟消息
        rabbitTemplate.convertAndSend(
                DelayRabbitConfig.DELAY_EXCHANGE,
                DelayRabbitConfig.DELAY_ROUTING_KEY,
                message,
                // 核心：设置单条消息延迟时间
                msg -> {
                    msg.getMessageProperties().setHeader("x-delay", delayTime);
                    return msg;
                }
        );
        return"延迟消息发送成功！预计 " + delayTime / 1000 + " 秒后执行，消息内容：" + message;
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
publicclassDelayMsgConsumer {
    /**
     * 监听延迟队列，手动ACK模式
     */
    @RabbitListener(queues = DelayRabbitConfig.DELAY_QUEUE)
    publicvoidconsumeDelayMessage(String msg, Message message, Channel channel)throws IOException {
        // 获取消息唯一标识
        longdeliveryTag= message.getMessageProperties().getDeliveryTag();
        try {
            // 执行业务逻辑
            System.out.println("【延迟消息消费成功】时间：" + System.currentTimeMillis() + "，消息内容：" + msg);
            // 手动确认消费成功，删除消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 消费异常，拒绝消息，重回队列重试
            System.err.println("【延迟消息消费失败】异常信息：" + e.getMessage());
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

http://localhost:8080/send/delay/message?message=订单超时自动取消&delayTime=10000