SpringBoot+RocketMQ实现延迟消息的示例代码

2025-10-25 10:31 开发作者：匆匆忙忙游刃有余

一、延迟消息概述

RocketMQ提供了两种类型的延迟消息机制：

延迟消息：消息发送后延迟指定的时间长度再被消费
定时消息：消息在指定的具体时间点被消费

这两种机制在订单超时取消、会议提醒、定时任务调度等场景中有广泛应用。

二、环境准备

1. 添加Maven依赖

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

2. 配置文件设置

在application.yml中配置RocketMQ连接信息：

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

三、延迟级别机制实现

1. 默认延迟级别

RocketMQ默认提供18个延迟级别，定义在MessageStoreConfig类中：

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

对应关系：

level=1: 延迟1秒
level=2: 延迟5秒
level=3: 延迟10秒
level=4: 延迟30秒
level=5: 延迟1分钟
level=6: 延迟2分钟
...以此类推
level=18: 延迟2小时

2. 基于延迟级别的生产者实现

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;
    
    /**
     * 发送基于延迟级别的消息
     * @param topic 主题
     * @param tag 标签
     * @param message 消息内容
     * @param delayLevel 延迟级别(1-18)
     */
    public void sendMessageByDelayLevel(String topic, String tag, String message, int delayLevel) {
        // 创建消息
        Message<String> springMessage = MessageBuilder.withPayload(message).build();
        
        // 发送延迟消息
        SendResult sendResult = rocketMQTemplate.syncSend(
            topic + ":" + tag, 
            springMessage, 
            3000, // 超时时间
            delayLevel // 延迟级别
        );
        
        System.out.println("延迟级别消息发送成功: " + sendResult);
    }
    
    /**
     * 发送订单超时取消消息（延迟15分钟）
     */
    public void sendOrderTimeoutMessage(String orderId) {
        String message = "订单超时取消: " + orderId;
        // 15分钟对应level=14（根据默认配置）
        sendMessageByDelayLevel("OrderTopic", "Timeout", message, 14);
    }
}

四、基于具体php时间的延迟消息实现

1. 定时消息生产者

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.MessageBuildepythonr;
import org.springframework.stereotype.Component;

import Java.util.Date;

@Component
public class ScheduledMessageProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    /**
     * 发送延迟指定毫秒数的消息
     */
    public void sendMessageWithDelayMs(String topic, String message, long delayMs) {
        // 计算投递时间
        long deliverTimeMs = System.currentTimeMillis() + delayMs;
        
        // 创建消息并设置投递时间
        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)python;
        System.out.println("延迟毫秒消息发送成功: " + sendResult);
    }
    
    /**
     * 发送指定时间点投递的消息
     */
    public void sendMessageAtTime(String topic, String message, Date deliverTime) {
        long deliverTimeMs = deliverTime.getTime();
        
        // 创建消息并设置投递时间
        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("定时投递消息发送成功: " + sendResult);
    }
    
    /**
     * 发送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.LocalDateTimphpe;
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) {
   yBWiRqLyam     String now = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        System.out.println("[" + now + "] 接收到订单超时消息: " + message);
        
        // 处理订单取消逻辑
        processOrderCancellation(message);
    }
    
    private void processOrderCancellation(String message) {
        // 提取订单ID
        String orderId = message.substring(message.indexOf(":") + 2);
        System.out.println("执行订单取消操作，订单ID: " + orderId);
        // 这里可以调用订单服务进行取消操作
    }
}

六、Controller层实现

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;
    
    /**
     * 发送基于延迟级别的消息
     */
    @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 "延迟级别消息发送成功，延迟级别: " + delayLevel;
    }
    
    /**
     * 发送订单超时取消消息
     */
    @PostMapping("/order/timeout")
    public String sendOrderTimeout(@RequestParam String orderId) {
        delayLevelProducer.sendOrderTimeoutMessage(orderId);
        return "订单超时取消消息已发送，订单ID: " + orderId;
    }
    
    /**
     * 发送延迟指定毫秒的消息
     */
    @PostMapping("/milliseconds")
    public String sendByDelayMs(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam long delayMs) {
        
        scheduledMessageProducer.sendMessageWithDelayMs(topic, message, delayMs);
        return "延迟毫秒消息发送成功，延迟: " + delayMs + "ms";
    }
    
    /**
     * 发送指定时间点的消息
     */
    @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 "定时消息发送成功，投递时间: " + deliverTime;
    }
}

七、自定义延迟级别配置

在Broker的配置文件中可以自定义延迟级别：

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

重启Broker使其生效。注意，修改延迟级别后，所有使用延迟级别的消息都会使用新的配置。

八、两种实现方式对比

特性	基于延迟级别	基于具体时间
灵活性	较低，只能使用预定义级别	高，可以精确到毫秒
适用版本	全版本支持	RocketMQ 5.x及以上版本完整支持
使用场景	固定延迟时间的场景	需要精确控制投递时间的场景
配置复杂度	简单，无需额外配置	可能需要在Broker端开启相关功能

九、使用注意事项

延迟精度：
- 延迟消息的投递时间不是完全精确的，有一定误差
- 在高并发场景下，误差可能会增大
版本兼容性：
- 基于具体时间的延迟消息在RocketMQ 5.x版本支持更完善
- 在低版本中可能需要使用延迟级别机制
性能考虑：
- 大量延迟消息可能会增加Broker的负担
- 对于长时间延迟的消息，考虑使用其他方案（如定时任务+消息队列组合）
消息可靠性：
- 延迟消息同样支持持久化，确保Broker重启后不会丢失
- 建议开启消息确认机制确保消息可靠投递

十、测试示例

发送订单超时取消消息（延迟15分钟）：
```
POST /api/delay/order/timeout?orderId=ORDER123456
```

发送10秒后投递的消息：

POST /api/delay/milliseconds?topic=TestTopic&message=HelloDelay&delayMs=10000

发送指定时间点的消息：

POST /api/delay/scheduled?topic=TestTopic&message=HelloScheduled&deliverTime=2024-12-25%2000:00:00

通过以上配置和代码，您可以在SpringBoot项目中轻松实现基于RocketMQ的延迟消息功能，满足各种定时任务和延迟处理的业务需求。

到此这篇关于SpringBoot+RocketMQ实现延迟消息的示例代码的文章就介绍到这了,更多相关SpringBoot RocketMQ 延迟内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！

继续阅读：SpringBoot RocketMQ 延时 SpringBoot RocketMQ 延迟

SpringBoot+RocketMQ实现延迟消息的示例代码

目录

一、延迟消息概述

二、环境准备

1. 添加Maven依赖

2. 配置文件设置

三、延迟级别机制实现

1. 默认延迟级别

2. 基于延迟级别的生产者实现

四、基于具体php时间的延迟消息实现

1. 定时消息生产者

五、消费者实现

六、Controller层实现

七、自定义延迟级别配置

八、两种实现方式对比

九、使用注意事项

十、测试示例

更多精彩内容

精彩评论

最新开发

基于C++实现一个Windows多功能计算器

C#实现高效读写Excel工作表

SpringBoot登录认证前后端实现方案：SpringBoot + Mybatis + JWT（图文实例）

Java 8 Optional解决空指针问题全面解析

nacos未自动注册问题及解决

开发排行榜

springboot后端存储富文本内容的思路与步骤(含图片内容)

PyCharm运行python测试,报错“没有发现测试”/“空套件”的解决

return base64.b64encode(b).decode(

基于C语言实现钻石棋游戏的示例代码

Sublime Text 3解决中文乱码问题（实测可用）