RocketMQ 延时级别配置方式-6ren

RocketMQ 延时级别配置方式

转载作者：qq735679552 更新时间：2022-09-29 22:32:09

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章RocketMQ 延时级别配置方式由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

RocketMQ 支持定时消息，但是不支持任意时间精度，仅支持特定的 level，例如定时 5s， 10s， 1m 等.

其中，level=0 级表示不延时，level=1 表示 1 级延时，level=2 表示 2 级延时，以此类推.

如何配置：

在服务器端（rocketmq-broker端）的属性配置文件中加入以下行:

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

描述了各级别与延时时间的对应映射关系.

这个配置项配置了从1级开始各级延时的时间，如1表示延时1s，2表示延时5s，14表示延时10m，可以修改这个指定级别的延时时间；。

时间单位支持：s、m、h、d，分别表示秒、分、时、天；。

默认值就是上面声明的，可手工调整；。

默认值已经够用，不建议调整【仅供参考，还是根据实际需要调整。调整默认值时注意同时要修改时间对应的level级别的值】。

如何发送延时消息：

发送延时消息只需要在客户端（rocketmq-client端）待发送的消息（ com.alibaba.rocketmq.common.message.Message ）中设置延时级别delayLevel即可.

 
    ? 
   
         Message msg =  
         new 
         Message(topicName, 
         "" 
         ,keys,message.getBytes()); 
        
         msg.setDelayTimeLevel(delayLevel); 
        
         SendResult sendResult = getMQProducer.send(msg);

RocketMQ定时(延迟)消息

RocketMQ 不支持任意时间自定义的延迟消息，仅支持内置预设值的延迟时间间隔的延迟消息.

预设值的延迟时间间隔为:

1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h 。

延时消息的使用场景

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存.

生产。

 
    ? 
   
         package 
         com.xin.rocketmq.demo.testrun; 
        
         import 
         com.xin.rocketmq.demo.config.JmsConfig; 
        
         import 
         org.apache.rocketmq.client.producer.DefaultMQProducer; 
        
         import 
         org.apache.rocketmq.client.producer.SendResult; 
        
         import 
         org.apache.rocketmq.common.message.Message; 
        
         public 
         class 
         ProducerDelay { 
        
         public 
         static 
         void 
         main(String[] args)  
         throws 
         Exception { 
        
         DefaultMQProducer producer =  
         new 
         DefaultMQProducer( 
         "please_rename_unique_group_name" 
         ); 
        
         producer.setNamesrvAddr( 
         "192.168.10.11:9876" 
         ); 
        
         producer.start(); 
        
         Message msg1 =  
         new 
         Message( 
        
         JmsConfig.TOPIC, 
        
         "订单001" 
         .getBytes()); 
        
         msg1.setDelayTimeLevel( 
         2 
         ); 
         //延迟5秒 
        
         Message msg2 =  
         new 
         Message( 
        
         JmsConfig.TOPIC, 
        
         "订单001" 
         .getBytes()); 
        
         msg2.setDelayTimeLevel( 
         4 
         ); 
         //延迟30秒 
        
         SendResult sendResult1 = producer.send(msg1); 
        
         SendResult sendResult2 = producer.send(msg2); 
        
         System.out.println( 
         "Product1-同步发送-Product信息={}" 
         + sendResult1); 
        
         System.out.println( 
         "Product2-同步发送-Product信息={}" 
         + sendResult2); 
        
         producer.shutdown(); 
        
         } 
        
         }

消费。

 
    ? 
   
         package 
         com.xin.rocketmq.demo.testrun; 
        
         import 
         com.xin.rocketmq.demo.config.JmsConfig; 
        
         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 
         ConsumerDelay { 
        
         public 
         static 
         void 
         main(String[] args)  
         throws 
         Exception { 
        
         // 实例化消费者 
        
         DefaultMQPushConsumer consumer =  
         new 
         DefaultMQPushConsumer( 
         "please_rename_unique_group_name" 
         ); 
        
         // 设置NameServer的地址 
        
         consumer.setNamesrvAddr( 
         "192.168.10.11:9876" 
         ); 
        
         // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息 
        
         consumer.subscribe(JmsConfig.TOPIC,  
         "*" 
         ); 
        
         // 注册消息监听者 
        
         consumer.registerMessageListener( 
         new 
         MessageListenerConcurrently() { 
        
         @Override 
        
         public 
         ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) { 
        
         for 
         (MessageExt message : messages) { 
        
         // Print approximate delay time period 
        
         System.out.println( 
         "Receive message[msgId=" 
         + message.getMsgId() +  
         "] " 
         + (System.currentTimeMillis() - message.getStoreTimestamp()) +  
         "ms later" 
         ); 
        
         } 
        
         return 
         ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 
        
         } 
        
         }); 
        
         // 启动消费者 
        
         consumer.start(); 
        
         } 
        
         }

以上为个人经验，希望能给大家一个参考，也希望大家多多支持我.

原文链接：https://blog.csdn.net/weixin_38951207/article/details/79022875 。

最后此篇关于RocketMQ 延时级别配置方式的文章就讲到这里了,如果你想了解更多关于RocketMQ 延时级别配置方式的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！。

文章推荐： SpringBoot项目集成日志的实现方法

文章推荐：使用res:bean属性复制避免null值覆盖版本

文章推荐： Eclipse+Maven构建Hadoop项目的方法步骤

文章推荐： Java @Async注解导致spring启动失败解决方案详解

c - 在英特尔 x64 汇编中执行函数的最快(CPU 方式)方式？
我一直在阅读有关汇编函数的内容，但对于是使用进入和退出还是仅使用调用/返回指令来快速执行，我感到很困惑。一种方式快而另一种方式更小吗？例如，在不内联函数的情况下，在汇编中执行此操作的最快(stdcal
解析元组的 Pythonic 方式
我正在处理一个元组列表，如下所示: res = [('stori', 'JJ'), ('man', 'NN'), ('unnatur', 'JJ'), ('feel', 'NN'), ('pig',
使用线程的 Pythonic 方式
最近我一直在做很多网络或 IO 绑定(bind)操作，使用线程有助于加快代码速度。我注意到我一直在一遍又一遍地编写这样的代码: threads = [] for machine, user, data
请求资源子集的 RESTful 方式
假设我有一个名为 user_stats 的资源，其中包含用户拥有的帖子、评论、喜欢和关注者的数量。是否有一种 RESTful 方式只询问该统计数据的一部分(即，对于 user_stats/3，请告诉我
接受选择的 RESTful 方式
我有一个简单的 api，它的工作原理是这样的: 用户创建一个请求 ( POST /requests ) 另一个用户检索所有请求 ( GET /requests ) 然后向请求添加报价 ( POST /
扩展构建器模式的 Pythonic 方式
考虑以下 CDK Python 中的示例(对于这个问题，不需要 AWS 知识，这应该对基本上任何构建器模式都有效，我只是在这个示例中使用 CDK，因为我使用这个库遇到了这个问题。): from aws
用于管理对象池的 Scala 方式
Scala 中管理对象池的首选方法是什么？我需要单线程创建和删除大规模对象(不需要同步)。在 C++ 中，我使用了静态对象数组。在 Scala 中处理它的惯用和有效方法是什么？最佳答案我会把它
将多个类方法应用于对象列表的 Pythonic 方式
我有一个带有一些内置方法的类。这是该类的抽象示例: class Foo: def __init__(self): self.a = 0 self.b = 0
返回和检查方法执行的 Pythonic 方式
返回和检查方法执行的 Pythonic 方式我目前在 python 代码中使用 golang 编码风格，决定移动 pythonic 方式例子: import sys from typing imp
处理丢失数据的 RESTful 方式
我正在开发一个 RESTful API。其中一个 URL 允许调用者通过 id 请求特定人员的记录。返回该 id 不存在的记录的常规值是什么？服务器是否应该发回一个空对象或者一个 404，或者其他什
加载具有相似文件名的文件的 Pythonic 方式
我正在使用 pathlib.Path() 检查文件是否存在，并使用 rasterio 将其作为图像打开. filename = pathlib.Path("./my_file-name.tif") 但
创建嵌套字典的 Pythonic 方式
我正在寻找一种 Pythonic 方式来从列表和字典创建嵌套字典。以下两个语句产生相同的结果: a = [3, 4] b = {'a': 1, 'b': 2} c = dict(zip(b, a))
处理异常后恢复的 Pythonic 方式
我有一个正在操裁剪理设备的脚本。设备有时会发生物理故障，当它发生时，我想重置设备并继续执行脚本。我有这个: while True: do_device_control() device
做组合别名的 Pythonic 方式
做组合别名的最pythonic和正确的方法是什么？这是一个假设的场景: class House: def cleanup(self, arg1, arg2, kwarg1=False):
在一个请求中创建多个项目的 RESTful 方式
我正在开发一个小型客户端服务器程序来收集订单。我想以“REST(ful)方式”来做到这一点。我想做的是: 收集所有订单行(产品和数量)并将完整订单发送到服务器目前我看到有两种选择: 将每个订单行发
动态调用静态方法的 Groovy 方式
我知道在 Groovy 中您可以使用字符串调用类/对象上的方法。例如: Foo."get"(1) /* or */ String meth = "get" Foo."$meth"(1) 有没有办法
复制没有不同属性的字典的 Pythonic 方式
在 ECMAScript6 中，您可以使用扩展运算符来解构这样的对象 const {a, ...rest} = obj; 它将 obj 浅拷贝到 rest，不带属性 a。有没有一种干净的方法可以在
函数中条件返回的 Pythonic 方式
我有几个函数返回数字或None。我希望我的包装函数返回第一个不是 None 的结果。除了下面的方法之外，还有其他方法吗？ def func1(): return None def func2(
处理资源之间双向关系的 RESTful 方式
假设我想设计一个 REST api 来讨论歌曲、专辑和艺术家(实际上我就是这样做的，就像我之前的 1312414 个人一样)。歌曲资源始终与其所属专辑相关联。相反，专辑资源与其包含的所有歌曲相关联。
有条件地使用上下文管理器的 Pythonic 方式
这是我认为必须经常出现的问题，但我一直无法找到一个好的解决方案。假设我有一个函数，它可以作为参数传递一个开放资源(如文件或数据库连接对象)，或者需要自己创建一个。如果函数需要自己打开文件，最佳实践通常

qq735679552

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城