跳到主要内容

RocketMQ 定时任务

RocketMQ 是一个分布式消息中间件，广泛应用于异步通信、解耦系统、流量削峰等场景。除了基本的消息发送和消费功能外，RocketMQ 还提供了强大的定时任务功能，允许开发者发送延迟消息，从而实现定时任务调度。

什么是RocketMQ定时任务？​

RocketMQ 的定时任务功能允许消息在指定的延迟时间后被消费。这种功能非常适合需要延迟处理任务的场景，例如订单超时取消、定时提醒、任务调度等。

RocketMQ 的定时任务是通过延迟消息实现的。发送消息时，可以指定一个延迟级别（delay level），消息会在指定的延迟时间后被投递到消费者。

延迟级别​

RocketMQ 提供了18个预定义的延迟级别，每个级别对应一个固定的延迟时间。以下是 RocketMQ 的延迟级别及其对应的延迟时间：

延迟级别延迟时间
11秒
25秒
310秒
430秒
51分钟
62分钟
73分钟
84分钟
95分钟
106分钟
117分钟
128分钟
139分钟
1410分钟
1520分钟
1630分钟
171小时
182小时
备注

RocketMQ 的延迟级别是固定的，无法自定义延迟时间。如果需要更灵活的延迟时间，可以考虑使用其他调度工具或自行实现。

如何使用定时任务？​

1. 发送延迟消息​

在发送消息时，可以通过设置 delayTimeLevel 属性来指定延迟级别。以下是一个 Java 示例：

java
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class ScheduledMessageProducer {
    public static void main(String[] args) throws Exception {
        // 实例化一个生产者
        DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息，指定Topic和消息体
        Message message = new Message("TestTopic", "Hello, RocketMQ!".getBytes());
        // 设置延迟级别为3，即延迟10秒
        message.setDelayTimeLevel(3);

        // 发送消息
        producer.send(message);
        System.out.println("消息已发送，将在10秒后投递");

        // 关闭生产者
        producer.shutdown();
    }
}

2. 消费延迟消息​

消费延迟消息与消费普通消息的方式相同。以下是一个 Java 消费者示例：

java
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class ScheduledMessageConsumer {
    public static void main(String[] args) throws Exception {
        // 实例化一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumerGroup");
        // 设置NameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅Topic
        consumer.subscribe("TestTopic", "*");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    System.out.println("收到消息: " + new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("消费者已启动，等待消息...");
    }
}

3. 运行结果​

运行上述生产者和消费者代码后，生产者会发送一条延迟10秒的消息。消费者将在10秒后收到并处理该消息。

实际应用场景​

1. 订单超时取消​

在电商系统中，用户下单后如果未在规定时间内支付，订单将被自动取消。可以使用 RocketMQ 的定时任务功能来实现这一需求。例如，用户下单后发送一条延迟30分钟的消息，如果30分钟后未支付，消费者将收到消息并执行取消订单的逻辑。

2. 定时提醒​

在任务管理系统中，可以为每个任务设置提醒时间。例如，用户设置了一个任务需要在1小时后提醒，系统可以发送一条延迟1小时的消息，消费者在收到消息后发送提醒通知。

3. 任务调度​

在分布式系统中，某些任务需要在特定时间执行。可以使用 RocketMQ 的定时任务功能来调度这些任务。例如，每天凌晨执行数据备份任务，可以发送一条延迟到凌晨的消息，消费者在收到消息后执行备份操作。

总结​

RocketMQ 的定时任务功能通过延迟消息实现，非常适合需要延迟处理任务的场景。通过设置延迟级别，可以轻松实现消息的延迟投递。在实际应用中，定时任务功能可以用于订单超时取消、定时提醒、任务调度等场景。

提示

如果你需要更灵活的延迟时间，可以考虑使用其他调度工具，如 Quartz、Spring Scheduler 等，或者自行实现延迟队列。

附加资源​

练习​

  1. 修改上述代码，发送一条延迟5分钟的消息，并验证消费者是否在5分钟后收到消息。
  2. 尝试实现一个订单超时取消的场景，用户下单后发送一条延迟30分钟的消息，如果30分钟后未支付，消费者将收到消息并打印“订单已取消”。