RocketMQ进阶

Stella981
• 阅读 637

点击上方 蓝字 关注我们

RocketMQ进阶

前言

在开发中经常会遇到延时任务的需求,例如在12306购买车票,若生成订单30分钟未支付则自动取消;还有在线商城完成订单后48小时不评价 ,自动5星好评。像这类在某事件触发后一段时间内执行的需求任务我们称之为 延时任务

那么如何实现延迟任务呢?

第一反应是利用cron方案来实现:RocketMQ进阶

启动一个cron定时任务,每隔一段时间执行一次,比如30分钟,找到那些超时的数据,直接更新状态,或者拿出来执行一些操作。如果数据量比较大,需要分页查询,分页update,这将是一个for循环更新操作。

cron方案是很常见的一种方案,但是常见的不一定是最好的,主要有以下几个问题:

  • 当数据量大的时候轮询效率低;

  • 时效性不够好,如果每小时轮询一次,最差的情况时间误差会达到1小时;

  • 如果通过增加cron轮询频率来减少时间误差,则会出现轮询低效和重复计算的问题;

既然cron方案不是很理想,那就请出我们今天的主角,使用RocketMQ的延时消息解决。在创建订单的时候发送一条延时消息到RocketMQ,30分钟后消费者消费消息去检查订单的状态,如果发现订单未支付则取消订单释放库存。

实现

RocketMQ延迟队列的核心思路是:所有的延迟消息由producer发出之后,都会存放到同一个topic(SCHEDULE_TOPIC_XXXX)下,不同的延迟级别会对应不同的队列序号,当延迟时间到之后,由定时线程读取转换为普通的消息存的真实指定的topic下,此时对于consumer端此消息才可见,从而被consumer消费。

注意:RocketMQ不支持任意时间的延时,只支持以下几个固定的延时等级
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

下面我们结合SprintBoot利用RocketMQ发送延时消息

  • 引入RocketMQ组件

     org.apache.rocketmq rocketmq-spring-boot-starter

  • 增加RocketMQ的配置

    rocketmq:  name-server: 172.31.0.44:9876  producer:    group: delay-group

  • 编写生产者

    @Component@Slf4jpublic class DelayProduce {    @Autowired    private RocketMQTemplate rocketMQTemplatet;    public void sendDelayMessage(String topic,String message,int delayLevel){       SendResult sendResult = rocketMQTemplatet.syncSend(topic, MessageBuilder.withPayload(message).build(), 2000, delayLevel);        log.info("sendtime is {}", DateTimeFormatter.ofPattern("yyyy年MM月dd日 HH:mm:ss").format(LocalDateTime.now()));        log.info("sendResult is{}",sendResult);    }}

  • 编写消费者

    @Slf4j@Component@RocketMQMessageListener(        topic = "delay-topic",        consumerGroup = "delay-group")public class DelayConsumer implements RocketMQListener {    @Override    public void onMessage(String message) {        log.info("received message time is {}", DateTimeFormatter.ofPattern("yyyy年MM月dd日 HH:mm:ss").format(LocalDateTime.now()));        log.info("received message is {}",message);    }}

  • 测试

    @RunWith(SpringRunner.class)@SpringBootTestpublic class DelayProduceTest {    @Autowired    private DelayProduce delayProduce;    @Test    public void sendDelayMessage() {        delayProduce.sendDelayMessage("delay-topic","Hello,JAVA日知录",5);    }}

这里delayLevel设置成5,对应RocketMQ的延时等级就是1分钟后投递消息。

  • 运行结果 RocketMQ进阶发送时间 RocketMQ进阶消费时间

修改延时级别

RocketMQ的延迟等级可以进行修改,以满足自己的业务需求,可以修改/添加新的level。例如:你想支持1天的延迟,修改最后一个level的值为1d,这个时候依然是18个level;也可以增加一个1d,这个时候总共就有19个level。

  • 打开RocketMQ的配置文件,修改 messageDelayLevel 属性

    brokerClusterName = DefaultClusterbrokerName = broker-abrokerId = 0deleteWhen = 04fileReservedTime = 48brokerRole = ASYNC_MASTERflushDiskType = ASYNC_FLUSHstorePathRootDir = /app/rocketmq/datamessageDelayLevel=90s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

这次将延时等级1修改成了90s,生产者发送消息后需要90s后再进行消息投递。修改完成后重启RocketMQ。nohup sh mqbroker -n localhost:9876 -c ../conf/broker.conf &

  • 使用延时等级1发送消息

    public void sendDelayMessage() { delayProduce.sendDelayMessage("delay-topic","Hello,JAVA日知录",1);}

  • 测试 RocketMQ进阶发送时间 RocketMQ进阶消费时间

通过比对发送时间与消费时间证明延时等级修改生效。

RocketMQ 相关文章

好了,各位朋友们,本期的内容到此就全部结束啦,能看到这里的同学都是优秀的同学,下一个升职加薪的就是你了!
如果觉得这篇文章对你有所帮助的话请扫描下面二维码加个关注。" 转发 " 加 " 在看 ",养成好习惯!咱们下期再见!

RocketMQ进阶

热文推荐

☞ 数据库优化之SQL优化
☞ 数据库优化之实例优化
☞ Docker基础与实战,看这一篇就够了!
☞ Docker-Compose基础与实战,看这一篇就够了!
OAuth2.0最简向导(多图预警)
☞ 构建三维一体立体化监控体系
☞ SpringCloud实战系列
☞ RocketMQ进阶-事务消息
☞ 数据库索引,你要了解的都在这里!

RocketMQ进阶

JAVA日知录

长按左边二维码关注我们,精彩文章第一时间推送

         >>>技术交流群<<<

朕已阅 RocketMQ进阶

本文分享自微信公众号 - JAVA日知录(javadaily)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
待兔 待兔
3个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
Android So动态加载 优雅实现与原理分析
背景:漫品Android客户端集成适配转换功能(基于目标识别(So库35M)和人脸识别库(5M)),导致apk体积50M左右,为优化客户端体验,决定实现So文件动态加载.!(https://oscimg.oschina.net/oscnet/00d1ff90e4b34869664fef59e3ec3fdd20b.png)点击上方“蓝字”关注我
Stella981 Stella981
3年前
Docker 部署SpringBoot项目不香吗?
  公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为★“星标”!这样才不会错过每日进阶架构文章呀。  !(http://dingyue.ws.126.net/2020/0920/b00fbfc7j00qgy5xy002kd200qo00hsg00it00cj.jpg)  2
Stella981 Stella981
3年前
200的大额人民币即将面世?央行:Yes!
点击上方蓝字关注我们!(https://oscimg.oschina.net/oscnet/2a1c2ac00bf54458a78c48a6c2e547d5.png)点击上方“印象python”,选择“星标”公众号重磅干货,第一时间送达!!(
可莉 可莉
3年前
200的大额人民币即将面世?央行:Yes!
点击上方蓝字关注我们!(https://oscimg.oschina.net/oscnet/2a1c2ac00bf54458a78c48a6c2e547d5.png)点击上方“印象python”,选择“星标”公众号重磅干货,第一时间送达!!(
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
9个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这