<!-- RabbitMQ AMQP 核心依賴(lài) -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    # 基礎(chǔ)連接配置
    host:127.0.0.1
    port:5672
    username:guest
    password:guest
    virtual-host:/
    # 開(kāi)啟生產(chǎn)者確認(rèn)
    publisher-confirm-type:correlated
    publisher-returns:true
    # 消費(fèi)者手動(dòng)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;
/**
 * 延遲消息隊(duì)列配置類(lèi)
 * 基于 rabbitmq-delayed-message-exchange 插件實(shí)現(xiàn)
 */
@Configuration
publicclassDelayRabbitConfig {
    // 延遲交換機(jī)名稱(chēng)
    publicstaticfinalStringDELAY_EXCHANGE="business_delay_exchange";
    // 延遲隊(duì)列名稱(chēng)
    publicstaticfinalStringDELAY_QUEUE="business_delay_queue";
    // 路由鍵
    publicstaticfinalStringDELAY_ROUTING_KEY="business.delay.routing";
    /**
     * 構(gòu)建延遲交換機(jī)
     * x-delayed-message：延遲交換機(jī)類(lèi)型
     * x-delayed-type：轉(zhuǎn)發(fā)模式（direct/topic/fanout）
     */
    @Bean
    public DirectExchange delayExchange() {
        Map<String, Object> args = newHashMap<>();
        // 核心參數(shù)：聲明為延遲交換機(jī)
        args.put("x-delayed-type", "direct");
        // 參數(shù)：名稱(chēng)、持久化、不自動(dòng)刪除、自定義參數(shù)
        returnnewDirectExchange(DELAY_EXCHANGE, true, false, args);
    }
    /**
     * 延遲隊(duì)列（持久化）
     */
    @Bean
    public Queue delayQueue() {
        returnnewQueue(DELAY_QUEUE, true);
    }
    /**
     * 隊(duì)列與延遲交換機(jī)綁定
     */
    @Bean
    public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {
        return BindingBuilder.bind(delayQueue)
                .to(delayExchange)
                .with(DELAY_ROUTING_KEY);
    }
    /**
     * 自定義RabbitTemplate，開(kāi)啟消息可靠投遞
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplaterabbitTemplate=newRabbitTemplate(connectionFactory);
        // 開(kāi)啟 mandatory，消息投遞失敗返回回調(diào)
        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;
    /**
     * 發(fā)送延遲消息接口
     * @param message 消息內(nèi)容
     * @param delayTime 延遲時(shí)間（單位：毫秒）
     * @return 結(jié)果
     */
    @GetMapping("/send/delay/message")
    public String sendDelayMessage(@RequestParam String message,
                                   @RequestParam Long delayTime) {
        // 發(fā)送延遲消息
        rabbitTemplate.convertAndSend(
                DelayRabbitConfig.DELAY_EXCHANGE,
                DelayRabbitConfig.DELAY_ROUTING_KEY,
                message,
                // 核心：設(shè)置單條消息延遲時(shí)間
                msg -> {
                    msg.getMessageProperties().setHeader("x-delay", delayTime);
                    return msg;
                }
        );
        return"延遲消息發(fā)送成功！預(yù)計(jì) " + delayTime / 1000 + " 秒后執(zhí)行，消息內(nèi)容：" + 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 {
    /**
     * 監(jiān)聽(tīng)延遲隊(duì)列，手動(dòng)ACK模式
     */
    @RabbitListener(queues = DelayRabbitConfig.DELAY_QUEUE)
    publicvoidconsumeDelayMessage(String msg, Message message, Channel channel)throws IOException {
        // 獲取消息唯一標(biāo)識(shí)
        longdeliveryTag= message.getMessageProperties().getDeliveryTag();
        try {
            // 執(zhí)行業(yè)務(wù)邏輯
            System.out.println("【延遲消息消費(fèi)成功】時(shí)間：" + System.currentTimeMillis() + "，消息內(nèi)容：" + msg);
            // 手動(dòng)確認(rèn)消費(fèi)成功，刪除消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 消費(fèi)異常，拒絕消息，重回隊(duì)列重試
            System.err.println("【延遲消息消費(fèi)失敗】異常信息：" + e.getMessage());
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

http://localhost:8080/send/delay/message?message=訂單超時(shí)自動(dòng)取消&delayTime=10000