扩展RocketMQ 使其支持任意时间精度的消息延迟

前言

本想使用rocketmq的延迟消息特性，但延迟的范围有限，仅支持
1s 5s 10s 30s
1m 2m 3m 4m
5m 6m 7m 8m
9m 10m 20m 30m
1h 2h
这18个等级，之外的延迟时间不支持，本文是为了说明如何利用rocketmq的延迟消息实现任意时间的消息延迟，如何进行扩展的

技术架构

rocketmq部署

下载安装包

https://mirror.bit.edu.cn/apache/rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip解压到目录 /Users/mengfanxiao/Documents/third_software/mq/rocketmq/rocketmq-all-4.7.0-bin-release

配置Java_home环境变量

sudo vim /etc/profile
jdk1.8安装报路径： /Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home添加 JAVA_HOME="/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home"
export JAVA_HOME
CLASS_PATH="$JAVA_HOME/lib"

安装nameserver

启动 mqnamesrv

sh bin/mqnamesrv

安装broker


sh bin/mqbroker -n 127.0.0.1:9876 autoCreateTopicEnable=true -c /Users/mengfanxiao/Documents/third_software/mq/rocketmq/rocketmq-all-4.7.0-bin-release/conf/broker.conf

安装console

下载 rocketmq 插件

代码下载

https://gitee.com/pingfanrenbiji/rocketmq-externals.git

修改配置

vim rocketmq-externals/rocketmq-console/src/main/resources/application.properties

代码编译

进入‘\rocketmq-externals\rocketmq-console’文件夹，执行‘mvn clean package -Dmaven.test.skip=true’，编译生成

启动

编译成功之后，Cmd进入‘target’文件夹，执行‘java -jar rocketmq-console-ng-1.0.1.jar’，启动‘rocketmq-console-ng-1.0.0.jar’。

访问控制台页面

吐槽下：页面和rabbitmq相比弱爆了

代码

producer

producer配置

producer 延迟消息生产流程解析

延迟发送逻辑

根据延迟时间和当前时间差值计算延迟等级

Integer level = DelayLevelCalculate.calculateDefault(l);

标记该消息为延迟消息希望该消息被消费的时候若发现是延迟消息即还未到消费的时间则不消费再次的触发生产者生产一次该消息

fillMessage(msg, level, startSendTime);

consumer

consumer配置

consumer 延迟消息消费流程解析

消息消费逻辑

若发现是延迟消息即还未到消费的时间 则不消费 再次的触发生产者生产一次该消息

参考代码

https://gitee.com/pingfanrenbiji/civism-rocket.git

参考文献

https://www.jianshu.com/p/4f968cd96b87

本文使用 mdnice 排版

扩展RocketMQ 使其支持任意时间精度的消息延迟相关推荐

扩展iQuery使其支持多种编程语言（二） – 兼编译器的语法分析简介
iQuery是一个开源的自动化测试框架项目,有兴趣的朋友可以在这里下载:https://github.com/vowei/iQuery/downloads 源码位置:https://github.co ...
扩展Snackbar 使其支持居中显示
https://github.com/nispok/snackbar 默认Snackbar支持底部或者顶部显示,不支持居中显示查看Snackbar.java的源码可以看到createMarginLa ...
扩展DailyRollingFileAppender使其支持限定日志文件的数量
本文用于备忘直接上代码: package hirebigdata.spider.linkedin.utils;import org.apache.log4j.DailyRollingFileAppe ...
AI智慧安监EasyCVR视频融合平台告警列表支持按时间查询告警消息
EasyCVR视频融合平台基于云边端架构,可支持海量视频汇聚管理,可提供视频监控直播.云端录像.云存储.录像检索与回看.智能告警.平台级联.智能分析等视频服务. 当前端设备包含报警布防功能时,用户通过 ...
扩展activiti 支持任意属性扩展
本文主要讲解如何扩展activiti,使其支持任意属性配置. 先来看下效果这个流程的分支二这个人员设置,实际存放xml格式如下: <userTask id="Activity_03o ...
kafka消息消费有延迟_如何在MQ中实现支持任意延迟的消息？
什么是定时消息和延迟消息? 定时消息:Producer 将消息发送到 MQ 服务端,但并不期望这条消息立马投递,而是推迟到在当前时间点之后的某一个时间投递到 Consumer 进行消费,该消息即定时消 ...
【RocketMQ】延迟消息(延迟队列)
文章目录 1. 什么是延迟消息 1.1 延时消息的使用场景 2. 示例 3. 原理参考 1. 什么是延迟消息发送消息后,消费者要等待一定的时间才能消费到该消息. RocketMQ 不支持任意时间自 ...
RocketMQ 任意时间维度的延迟消息（秒级）
RocketMQ 任意时间维度的延迟消息(秒级) 基于开源版本固定等级的延迟消息实现原理的基础上进行扩展,将所有维度的延迟消息封装成任务添加到时间轮上,通过时间轮固定周期的扫描,检测任务是否到执行时间 ...
阿里开发者招聘节 | 面试题06-07： MySQL的数据如何恢复到任意时间点
为帮助开发者们提升面试技能.有机会入职阿里,云栖社区特别制作了这个专辑--阿里巴巴资深技术专家们结合多年的工作.面试经验总结提炼而成的面试真题这一次将陆续放出(面试题官方参考答案将在专辑结束后统一汇总 ...

扩展RocketMQ 使其支持任意时间精度的消息延迟

前言

技术架构

rocketmq部署

下载安装包

配置Java_home环境变量

安装nameserver

安装broker

安装console

代码

producer

consumer

参考代码

参考文献

扩展RocketMQ 使其支持任意时间精度的消息延迟相关推荐

最新文章

热门文章