<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

rocketmq:
  name-server: localhost:9876
  producer:
    group: delay-message-producer-group

messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class DelayLevelProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    /**
     * 發(fā)送基于延遲級(jí)別的消息
     * @param topic 主題
     * @param tag 標(biāo)簽
     * @param message 消息內(nèi)容
     * @param delayLevel 延遲級(jí)別(1-18)
     */
    public void sendMessageByDelayLevel(String topic, String tag, String message, int delayLevel) {
        // 創(chuàng)建消息
        Message<String> springMessage = MessageBuilder.withPayload(message).build();
        
        // 發(fā)送延遲消息
        SendResult sendResult = rocketMQTemplate.syncSend(
            topic + ":" + tag, 
            springMessage, 
            3000, // 超時(shí)時(shí)間
            delayLevel // 延遲級(jí)別
        );
        
        System.out.println("延遲級(jí)別消息發(fā)送成功: " + sendResult);
    }
    
    /**
     * 發(fā)送訂單超時(shí)取消消息（延遲15分鐘）
     */
    public void sendOrderTimeoutMessage(String orderId) {
        String message = "訂單超時(shí)取消: " + orderId;
        // 15分鐘對(duì)應(yīng)level=14（根據(jù)默認(rèn)配置）
        sendMessageByDelayLevel("OrderTopic", "Timeout", message, 14);
    }
}

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Date;

@Component
public class ScheduledMessageProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    /**
     * 發(fā)送延遲指定毫秒數(shù)的消息
     */
    public void sendMessageWithDelayMs(String topic, String message, long delayMs) {
        // 計(jì)算投遞時(shí)間
        long deliverTimeMs = System.currentTimeMillis() + delayMs;
        
        // 創(chuàng)建消息并設(shè)置投遞時(shí)間
        Message<String> springMessage = MessageBuilder.withPayload(message)
            .setHeader(MessageConst.PROPERTY_DELAY_TIME_MS, String.valueOf(delayMs))
            .setHeader(MessageConst.PROPERTY_TIMER_DELIVER_MS, String.valueOf(deliverTimeMs))
            .build();
        
        SendResult sendResult = rocketMQTemplate.syncSend(topic, springMessage);
        System.out.println("延遲毫秒消息發(fā)送成功: " + sendResult);
    }
    
    /**
     * 發(fā)送指定時(shí)間點(diǎn)投遞的消息
     */
    public void sendMessageAtTime(String topic, String message, Date deliverTime) {
        long deliverTimeMs = deliverTime.getTime();
        
        // 創(chuàng)建消息并設(shè)置投遞時(shí)間
        Message<String> springMessage = MessageBuilder.withPayload(message)
            .setHeader(MessageConst.PROPERTY_TIMER_DELIVER_MS, String.valueOf(deliverTimeMs))
            .build();
        
        SendResult sendResult = rocketMQTemplate.syncSend(topic, springMessage);
        System.out.println("定時(shí)投遞消息發(fā)送成功: " + sendResult);
    }
    
    /**
     * 發(fā)送10秒后投遞的消息
     */
    public void sendTenSecondsLaterMessage(String topic, String message) {
        sendMessageWithDelayMs(topic, message, 10000L);
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Component
@RocketMQMessageListener(
    topic = "OrderTopic",
    consumerGroup = "delay-message-consumer-group",
    selectorExpression = "Timeout"
)
public class OrderTimeoutConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        String now = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        System.out.println("[" + now + "] 接收到訂單超時(shí)消息: " + message);
        
        // 處理訂單取消邏輯
        processOrderCancellation(message);
    }
    
    private void processOrderCancellation(String message) {
        // 提取訂單ID
        String orderId = message.substring(message.indexOf(":") + 2);
        System.out.println("執(zhí)行訂單取消操作，訂單ID: " + orderId);
        // 這里可以調(diào)用訂單服務(wù)進(jìn)行取消操作
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.web.bind.annotation.*;

import java.util.Date;

@RestController
@RequestMapping("/api/delay")
public class DelayMessageController {
    
    @Autowired
    private DelayLevelProducer delayLevelProducer;
    
    @Autowired
    private ScheduledMessageProducer scheduledMessageProducer;
    
    /**
     * 發(fā)送基于延遲級(jí)別的消息
     */
    @PostMapping("/level")
    public String sendByDelayLevel(
            @RequestParam String topic,
            @RequestParam String tag,
            @RequestParam String message,
            @RequestParam(defaultValue = "3") int delayLevel) {
        
        delayLevelProducer.sendMessageByDelayLevel(topic, tag, message, delayLevel);
        return "延遲級(jí)別消息發(fā)送成功，延遲級(jí)別: " + delayLevel;
    }
    
    /**
     * 發(fā)送訂單超時(shí)取消消息
     */
    @PostMapping("/order/timeout")
    public String sendOrderTimeout(@RequestParam String orderId) {
        delayLevelProducer.sendOrderTimeoutMessage(orderId);
        return "訂單超時(shí)取消消息已發(fā)送，訂單ID: " + orderId;
    }
    
    /**
     * 發(fā)送延遲指定毫秒的消息
     */
    @PostMapping("/milliseconds")
    public String sendByDelayMs(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam long delayMs) {
        
        scheduledMessageProducer.sendMessageWithDelayMs(topic, message, delayMs);
        return "延遲毫秒消息發(fā)送成功，延遲: " + delayMs + "ms";
    }
    
    /**
     * 發(fā)送指定時(shí)間點(diǎn)的消息
     */
    @PostMapping("/scheduled")
    public String sendScheduled(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") Date deliverTime) {
        
        scheduledMessageProducer.sendMessageAtTime(topic, message, deliverTime);
        return "定時(shí)消息發(fā)送成功，投遞時(shí)間: " + deliverTime;
    }
}

# 在broker.conf文件中添加
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 3h 4h 5h