java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > SpringBoot RocketMQ 延迟

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

作者：匆匆忙忙游刃有余

本文主要介绍了SpringBoot+RocketMQ实现延迟消息案例详解,包括基于延迟级别和基于具体时间两种方式的完整实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

下面将详细介绍如何在SpringBoot中使用RocketMQ实现延迟消息，包括基于延迟级别和基于具体时间两种方式的完整实现。

一、延迟消息概述

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

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

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

二、环境准备

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"

对应关系：

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);
    }
}

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

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.MessageBuilder;
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);
        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.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 + "] 接收到订单超时消息: " + 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端开启相关功能

九、使用注意事项

  1. 延迟精度：

    • 延迟消息的投递时间不是完全精确的，有一定误差
    • 在高并发场景下，误差可能会增大

  2. 版本兼容性：

    • 基于具体时间的延迟消息在RocketMQ 5.x版本支持更完善
    • 在低版本中可能需要使用延迟级别机制

  3. 性能考虑：

    • 大量延迟消息可能会增加Broker的负担
    • 对于长时间延迟的消息，考虑使用其他方案（如定时任务+消息队列组合）

  4. 消息可靠性：

    • 延迟消息同样支持持久化，确保Broker重启后不会丢失
    • 建议开启消息确认机制确保消息可靠投递

十、测试示例

  1. 发送订单超时取消消息（延迟15分钟）：

    POST /api/delay/order/timeout?orderId=ORDER123456

  2. 发送10秒后投递的消息：

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

  3. 发送指定时间点的消息：

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

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

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

您可能感兴趣的文章:
阅读全文