RocketMQ进击

文章11 |   阅读 6136 |   点赞0

来源：https://blog.csdn.net/itanping/category_9355252.html

RocketMQ进击（四）定时消息（延时队列）

转x33g5p2x  于2021-12-19 转载在 其他  
字(5.2k)|赞(0)|评价(0)|浏览(497)

楔子：大字半边床，口水枕边流。早上七点的闹钟响起，啊，起床上班；可恶，大床把我抱住了，起不来，再让我睡10分钟吧。嗯，好吧，原来是个梦，我在高速服务区睡着了。前者是定时指令/消息，定好闹钟后，每天早上到点就会叫你起床；后者是延时指令/消息，它会延迟当前的事情到相对于现在之后的某个时间点再做。但不管是定时还是延时，他们都有一个共同点：大脑到了这个时间点，它就是触发并工作，让你起床去上班。因为其本质都是一种相对的延迟再做。

像这样的定时消息和延时消息经常会出现的我们的生活中：

  • 周一早上10点项目早会；周五同事们约好晚上六点去外面吃一顿；等等类似，是定时消息
  • 周二下午2点的会议推迟半小时再开；周四晚上六点的上线要延迟到22点；等等类似，是延时消息

1. 似而不同

  • 定时消息：Producer 将消息发送到消息队列 MQ 服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到 Consumer 进行消费，该消息即定时消息。
  • 延时消息：Producer 将消息发送到消息队列 MQ 服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到 Consumer 进行消费，该消息即延时消息。

定时消息与延时消息虽然在理解角度有一些差异，但是最终达到的效果相同：消息在发送到消息队列 MQ 服务端后并不会立马投递，而是根据消息中的属性延迟固定时间后才投递给消费者。因为其本质都是一种相对的延迟再消费。

2. 到点出场

定时消息/延时消息适用于以下一些场景：

  • 消息生产和消费有时间窗口要求：比如在电商交易中超时未支付关闭订单的场景，用户提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。
  • 通过消息触发一些定时任务，比如在某一固定时间点向用户发送提醒消息。

2.2.1 源码示例

定时/延时消息生产者（Producer ）

package com.meiwei.service.mq.tcp.producer;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

/**
 * 定时消息（延迟队列） - 生产者
 */
public class DelayTimeMqProducer {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";

    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_DELAY_TIME";

    public static void main(String[] args) throws Exception {
        // 声明并实例化一个 producer 生产者来产生消息
        // 需要一个 producer group 名字作为构造方法的参数
        DefaultMQProducer producer = new DefaultMQProducer("meiwei-producer-delay-time");

        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 在发送MQ消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();

        // 循环发送MQ测试消息
        String content = "";
        for (int i = 0; i < 3; i++) {
            // 配置容灾机制，防止当前消息异常时阻断发送流程
            try {
                content = "【MQ测试消息】延时消息发送 " + i + "，时间 " + new Date();

                // Message Body 可以是任何二进制形式的数据，消息队列不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                Message message = new Message(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH, "KEY" + i, content.getBytes(RemotingHelper.DEFAULT_CHARSET));

                // 默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level
                // 设置延时等级3，这个消息将在 10s 之后发送
                message.setDelayTimeLevel(3);
                // 发送消息
                SendResult sendResult = producer.send(message);

                // 日志打印
                System.out.printf("Send MQ message success! Topic: %s，Tag: %s, msgId: %s, Message: %s %n",
                        message.getTopic(), message.getTags(), sendResult.getMsgId(), new String(message.getBody()));
            } catch (Exception e) {
                // 消息发送失败
                System.out.printf("%-10d Exception %s %n", i, e);
                e.printStackTrace();
            }
        }

        // 在发送完消息之后，销毁 Producer 对象。如果不销毁也没有问题
        producer.shutdown();
    }
}

在应用层初始化 Message 消息对象之后，调用 Message.setDelayTimeLevel(int level) 方法来设置延迟级别，按照序列取相应的延迟级别，例如 level=3，则延迟为 10s 再发送消息。 

RocketMQ 目前只支持固定精度时间的延时消息发送（配置 Message.setDelayTimeLevel 延时精度），默认有18个时间精度的 level，分别是：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

  • 这个配置项配置了从1级开始，各级延时的时间，可以修改这个指定级别的延时时间
  • 时间单位支持：s、m、h、d，分别表示秒、分、时、天
  • 默认值就是上面声明的，可手工调整
  • 默认值已够用，不建议修改这个值

定时/延时消息消费者（Consumer ）【Push模式】

package com.meiwei.service.mq.tcp.consumer;

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.Calendar;
import java.util.List;

/**
 * 定时消息（延迟队列） - 消费者
 */
public class DelayTimeMqConsumer {

    // Message 所属的 Topic 一级分类，须要与提供者的频道保持一致才能消费到消息内容
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_DELAY_TIME";

    public static void main(String[] args) throws Exception {

        // 声明并初始化一个 consumer
        // 需要一个 consumer group 名字作为构造方法的参数
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("meiwei-consumer-delay-time");

        // 同样也要设置 NameServer 地址，须要与提供者的地址列表保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 设置 consumer 所订阅的 Topic 和 Tag，*代表全部的 Tag
        consumer.subscribe(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH);

        // 注册消息监听者
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.forEach(mq->{
                    Calendar calendar = Calendar.getInstance();
                    String timeNow = new String(calendar.get(Calendar.HOUR_OF_DAY) + ":" + calendar.get(Calendar.MINUTE) + ":" + Calendar.SECOND);
                    System.out.printf("TimeNow: %s, Thread: %s, Topic: %s, Tags: %s, Message: %s",
                            timeNow,
                            Thread.currentThread().getName(),
                            mq.getTopic(),
                            mq.getTags(),
                            new String(mq.getBody()));
                    System.out.println();
                });

                // 返回消费状态
                // CONSUME_SUCCESS 消费成功
                // RECONSUME_LATER 消费失败，需要稍后重新消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 调用 start() 方法启动 consumer
        consumer.start();
        System.out.println("DelayTime Consumer Started.");
    }
}

2.2.2 测试结果

定时/延时消息生产者（Producer）发送结果：

定时/延时消息消费者（Consumer）消费结果：

可以看到消息的消费比存储时间晚了10秒。 

参考资料：
RocketMQ 官网：http://rocketmq.apache.org/docs/motivation/
阿里云消息队列 MQ：https://help.aliyun.com/document_detail/29532.html
阿里巴巴中间件团队：http://jm.taobao.org/2016/11/29/apache-rocketmq-incubation/

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/itanping/article/details/101070156

内容来源于网络，如有侵权，请联系作者删除！

Javarocketmqmq延迟队列mq定时消息

相关文章

    查看更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com