RocketMQ延迟消息源码分析

news2024/10/7 6:41:14

写作目的

第一个原因:最近玩哔哩哔哩遇到一个RocketMQ的Contributor,一开始不知道他是Contributor,后来问到延迟消息的时候这块还不是很了解,他告诉我学习要系统,你既然了解事务消息那我理解应该也了解延迟消息,事实我不了解,所以这块想通过看源码的方式了解一下。
第二个原因:好久没写文章了,需要水一篇,也需要不断学习,所以搞一下。

源码分析

延迟消息配置

消息的延时级别level一共有18级,分别为:

 private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

在这里插入图片描述

延迟消息发送

生产延迟消息的代码如下

  public static void main(String[] args) throws Exception {
    // Instantiate with a producer group name.
    DefaultMQProducer producer = new DefaultMQProducer("producer_group_name");
    producer.setNamesrvAddr("localhost:9876");
    producer.start();
    System.out.println("----------------");

    String topic = "DelayDemo";
    Message msg =
            new Message(
                    topic /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ ")/* Message body */ 
                            .getBytes(RemotingHelper.DEFAULT_CHARSET) );
    //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    msg.setDelayTimeLevel(3);//下标有0开始

    SendResult sendResult = producer.send(msg, 1000000000);
    System.out.printf("%s%n", sendResult);
    // Shut down once the producer instance is not longer in use.
    producer.shutdown();
  }

核心的话就是设置延迟消息等级的参数

    //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    msg.setDelayTimeLevel(3);//下标有0开始

那么设置的这个参数的意思是什么呢?可以跟进源码看一下。原来是给这个消息设置了一个KV,仅仅是打一个tag,后面会用到。

 public void setDelayTimeLevel(int level) {
 //String PROPERTY_DELAY_TIME_LEVEL = "DELAY";
        this.putProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(level));
    }

broker消息接收

目的:我们要定位是到消息生产者Clinet生产的消息被broker接收到走的哪块源码。
首先我们要找到程序的入口代码,如下图所示,就是一个broker的main方法启动。
在这里插入图片描述
跟进start方法,然后一步一步往下跟,会跟到NettyRemotingServer的start方法
在这里插入图片描述
NettyRemotingServer的start方法里会就是Netty创建ServerBootstrap了,那么很自然的就会想到自定义的handler,也就是NettyServerHandler。
在这里插入图片描述
NettyServerHandler#channelRead0方法是处理接收到的程序,进一步跟到NettyRemotingAbstract#processRequestCommand方法,我们可以推断出根据消息的code找到具体的NettyRequestProcessor,就可以知道具体的消息存储逻辑了。
在这里插入图片描述

那么发现消息的code是什么呢?
消息生产者发送消息的时候一直跟源码,就会跟到下图的这个地方
在这里插入图片描述
就可以拿到

 public static final int SEND_MESSAGE_V2 = 310;

根据这个code就可以定位到处理消息的processor,即SendMessageProcessor
在这里插入图片描述

延迟消息存储到CommitLog

从上面的逻辑中我们已经定位到SendMessageProcessor,那么接下来看一下消息存储的粗略逻辑
从SendMessageProcessor#processRequest方法开始跟,如下图所示
在这里插入图片描述
最后跟到CommitLog的asyncPutMessage方法,其中里面有一个分支如下图所示

在这里插入图片描述
接下来就是正常的存储了

延迟消息构建Consumequeue

Consumequeue的构建在RocketMQ中msg&tag的生命周期4.2小节有讲过。
接下来看一下延迟消息构建过程。

核心在ReputMessageService#doReput方法里的构建DispatchRequest方法

 DispatchRequest dispatchRequest =
                                DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);

checkMessageAndReturnSize就是构建DispatchRequest ,当然也会构建tagsCode。如下面的代码所示,如果是延迟消息,则tagsCode=存储时间+延迟时间
在这里插入图片描述

延迟消息定时任务

源码剖析RocketMQ延时消息原理第3小节中讲的很详细。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/418698.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

BGP路由实验

要求1 使用 Preval 策略 [r4]ip ip-prefix PV permit 192.168.10.0 24 [r4]route-policy PV permit node 10 [r4-route-policy]if-match ip-prefix PV [r4-route-policy]apply preferred-value 100 [r4]route-policy PV permit node 20 [r4-bgp]peer 24.0.0.2 route-policy…

redis网络模型

用户空间和内核空间IO五种IO模型阻塞IO非阻塞IOIO多路复用selectpollepollweb服务流程信号驱动IO异步IOIO模型比较redis网络模型redis为什么是单线程redis单线程网络模型流程用户空间和内核空间 为安全,将用户应用和系统应用分隔开,产生用户空间和内核空…

thanos prometheus 的高可用、长期存储二进制部署

1.简介 http://thanos.io/ thanos 是具有长期存储功能的开源、高可用性 Prometheus的集群组件。 全局查询视图 跨多个 Prometheus 服务器和集群查询指标 无限保留 使用对象存储扩展系统,不限时间保留指标。 Prometheus兼容 兼容 Prometheus api,用于…

我在windows10下,使用msys64 mingw64终端

系列文章目录 文章目录系列文章目录前言一、MSYS2是什么?前言 msys2官网 MSYS2 (Minimal SYStem 2) 是一个MSYS的独立改写版本,主要用于 shell 命令行开发环境。 同时它也是一个在Cygwin (POSIX 兼容性层&#xff09…

软件测试简历如何编写?还在乱写?精细优化让自己脱颖而出......

目录:导读前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结(尾部小惊喜)前言 简历是我们向面试官…

教育大数据总体解决方案(6)

触控录播主机集成视频编码、实时导播、音频处理、图像跟踪、电子云台等功能,无需额外的辅助跟踪设备。支持5路视频信号的采集录制、画面自动跟踪及全自动/半自动导播切换,让教师免于分心调控,专注于课堂教学。 整机采取一体化设计及三合一按键…

【数据结构与算法】栈和队列(StackQueue)

TOC 数据结构—栈 栈的概念 要想学习一个东西,概念是一定要看并且理解的,那么栈是个什么玩意呢? 栈:一种特殊的线性表,其只允许在固定的一端进行插入和删除元素操作。进行数据插入和删除操作的一端称为栈顶&#x…

v851s g2d 模块 sample 深究

1. g2d 模块概述 g2d 主要功能: 1)旋转:支持90、180、270旋转; 2)镜像反转:H / V; 3) scale:放缩 4)格式转换:yuv 转 rgb 等,多种格式相互间转换; 5)透明叠加功能:实现两个rgb图片叠加; 6)矩形填充,等诸多功能; 2. g2d 配置 1)源码目录:tina-v853-docker/…

Zabbix自定义监控mysql数据库、自动注册服务器及部署代理服务器

目录 一、zabbix自定义监控数据库 1、编写监控脚本 2、服务端测试 3、web页面配置 ①创建自定义监控项 ②创建触发器 ③创建图形 ④测试自定义监控是否成功 二、zabbix自动注册 1、什么是自动注册 2、环境准备 3、 zabbix客户端配置 4、web页面配置自动注册 5、…

Windows系统生产力工具介绍

介绍 本文主要介绍在windows系统上如何安装一些常用的生产力软件,这些软件大多数都是开源免费使用的,包括markdown编辑器、知识管理软件、图片和视频工具、系统工具等,以及程序员专用的开发工具。根据本人的使用经验,将会不定期更…

OpenGL之深入解析屏幕成像和渲染原理

一、CPU 与 GPU CPU 内部组成:GPU 内部组成(ALU:算术逻辑单元,是能实现多组算术运算和逻辑运算的组合逻辑电路):CPU 和 GPU 因为设计之初需求就不一样,所以它们的组成不同,在计算机中的分工也不同。可以看到,GPU 有更多的 ALU,而 CPU 有 Control 单元和 Cache 单元,…

TensorFlow 深度学习第二版:6~10

原文:Deep Learning with TensorFlow Second Edition 协议:CC BY-NC-SA 4.0 译者:飞龙 本文来自【ApacheCN 深度学习 译文集】,采用译后编辑(MTPE)流程来尽可能提升效率。 不要担心自己的形象,只…

金融风险计量:数据平滑方法及逆平滑分析

摘要及声明 1:本文从风险分析的角度简单介绍数据平滑方式,重点介绍低频数据的逆平滑分析; 2:本文主要数据通过爬虫获取; 3:模型实现基于python3.8; 处理金融数据时我们经常会遇到有噪音的数…

规模化敏捷框架:Spotify

Spotify 是全球最大、最受欢迎的流媒体音乐服务平台,预估用户总量已达2.86亿。Spotify 取得成功的一个关键因素就在于公司采用了一个独特方法: 围绕工作任务进行组织构建以提高团队敏捷性。Spotify 工程团队把提高团队敏捷性的经验记录了下来,并把经验分…

Java web学习记录(二)数据库的使用

学习Java web的前置条件就是数据库,只有学了数据库才能更好的处理网站应用产生的数据。 初识数据库 数据库(Database)顾名思义就是一个存储数据的仓库,通过它就可以直接查找到你想要的数据,举个简单的例子&#xff0…

IPSEC实验(IPSECVPN点到点,DSVPN,IPSECVPN旁挂)

目录一、复现实验1、防火墙的IPSECVPN点到点实验-1,拓扑图的搭建-2,配置IP,开通ping,并且设置策略-3,在网络中的IPSEC进行配置第一阶段:发出的UDP500流量第二阶段 发出的ESP流量二台防火墙建立策略禁用其它策略,在IPSEC上配置策略…

系统应满足的性能指标计算及系统性能衡量

根据运营数据计算系统应该满足的性能指标 计算正常业务操作(稳定性测试)的并发量 计算峰值业务操作(压力测试)的并发量 如何进行并发数计算(稳定性测试和压力测试) 使用阶梯线程组自定义模拟用户数量。 …

前端开发工具-Visual Studio Code-插件下载-迁移到新电脑

背景 前端使用的开发工具一般是Visual Studio Code,很多辅助功能,比如字体高亮、单词拼写检查、预览图片等需要安装插件。但是插件在原来的电脑,不想下载或者自己是新人,想迁移同事的插件,或者新电脑没有外网。 以下…

手把手教学在windows系统上将pytorch模型转为onnx,再转为ncnn的全过程

前言 最近呢,在忙一个项目,需要将pytorch训练的模型部署在移动端。然后遇到也遇到了一些坑,简单的记录一下整个过程,转换的模型就使用经典的分类网络模型mobilenet_v2。 将pytorch模型转换为onnx模型 环境准备 这个步骤比较简…

U+平台和华为软开平台怎么拉取远程项目作为新项目

这是根据聊天记录改的帖子,这样应该算得上是一篇技术博客了吧,又完成一个指标【狗头】 用idea作为开发工具 首先连接校园网,然后进入U网址http://10.5.1.21:30080/student/group 从这进去 修改https密码,选择修改,不…