余晖落尽暮晚霞,黄昏迟暮远山寻
本站
当前位置:网站首页 > 编程知识 > 正文

作为Java程序员,你应该知道的RocketMQ工作原理

xiyangw 2023-05-14 11:30 12 浏览 0 评论

先简单了解一下各个模块的功能,下面会有章节详细介绍各个模块的功能。

作为Java程序员,你应该知道的RocketMQ工作原理

Producer和Consumer很好理解,顾名思义就是生产者和消费者,生产者负责生产消息,消费者负责消费消息,这2块的逻辑都是由业务使用者定义的。

Broker是RocketMQ的核心,Broker实现了消息的存储、拉取等功能。Broker通常以集群方式启动,并可配置主从,每个Broker上提供对指定topic的服务。理解了Broker的原理,以及和其他服务交互的方式就基本弄懂了整个消息中间件的原理。

NameSrv是一个无状态的名称服务,可以集群部署。所有Broker启动的时候会向NameSrv注册自己的信息。Producer会根据目标topic从NameSrv获取到达指定Broker的路由信息,Consumer同理。

对于Producer端RocketMQ采用了轮询的方式保证了负载均衡,Consumer端通常采用cluster集群方式消费消息,我们可以自己定义消息在消息端的分配方式。另外,MQ还提供了顺序消息的特性,简单了解一下MQ提供的特性即可,具体实现后面章节会进行阐述。

源码目录结构介绍&Remoting通信层

源码目录结构介绍 RocketMQ源码分为以下几个package

  • rocketmq-broker:整个mq的核心,他能够接受producer和consumer的请求,并调用store层服务对消息进行处理。HA服务的基本单元,支持同步双写,异步双写等模式。
  • rocketmq-client:mq客户端实现,目前官方仅仅开源了java版本的mq客户端,c++,go客户端有社区开源贡献。
  • rocketmq-common:一些模块间通用的功能类,比如一些配置文件、常量。
  • rocketmq-example:官方提供的例子,对典型的功能比如order message,push consumer,pull consumer的用法进行了示范。
  • rocketmq-filtersrv:消息过滤服务,相当于在broker和consumer中间加入了一个filter代理。
  • rocketmq-remoting:基于netty的底层通信实现,所有服务间的交互都基于此模块。
  • rocketmq-srvut:解析命令行的工具类。
  • rocketmq-store:存储层实现,同时包括了索引服务,高可用HA服务实现。
  • rocketmq-tools:mq集群管理工具,提供了消息查询等功能。

底层基于Netty网络库驱动

producer 1.启动流程

Producer如何感知要发送消息的broker即brokerAddrTable中的值是怎么获得的,

  1. 发送消息的时候指定会指定topic,如果producer集合中没有会根据指定topic到namesrv获取topic发布信息TopicPublishInfo,并放入本地集合
  2. 定时从namesrv更新topic路由信息,

Producer与broker间的心跳

Producer定时发送心跳将producer信息(其实就是procduer的group)定时发送到, brokerAddrTable集合中列出的broker上去

Producer发送消息只发送到master的broker机器,在通过broker的主从复制机制拷贝到broker的slave上去

producer 2.如何发送消息

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡

1. Topic下的所有队列如何理解

比如broker1, broker2,
borker3三台broker机器都配置了Topic_A
Broker1 的队列为queue0 , queue1
Broker2 的队列为queue0, queue2, queue3,
Broker3 的队列为queue0
当然一般情况下的broker的配置都是一样的
以上当broker启动的时候注册
到namesrv的Topic_A队列为共6个分别为:
broker1_queue0, broker1_queue1,
broker2_queue0, broker2_queue1, broker2_queue2,
broker3_queue0,

2. Producer如何实现轮询队列

Producer从namesrv获取的到Topic_A
路由信息TopicPublishInfo
--List<MessageQueue>messageQueueList
//Topic_A的所有的队列
--AtomicIntegersendWhichQueue //自增整型
方法selectOneMessageQueue方法
用来选择一个发送队列
(++sendWitchQueue)% messageQueueList.size
为队列集合的下标
每次获取queue都会通过sendWhichQueue
加一来实现对所有queue的轮询
如果入参lastBrokerName不为空,
代表上次选择的queue发送失败,
这次选择应该避开同一个queue

3. Producer发消息系统重试

发送失败后,重试几次retryTimesWhenSendFailed = 2
发送消息超时sendMsgTimeout = 3000
Producer通过selectOneMessageQueue
方法获取一个MessagQueue对象
--topic //Topic_A
--brokerName //代表发送消息到达的broker
--queueId //代表发送消息的在指定broker
上指定topic下的队列编号
向指定broker的指定topic的指定queue发送消息
发送失败(1)重试次数不到两次(2)
发送此条消息花费时间还没有到3000(毫秒),
换个队列继续发送。

producer发送普通消息

producer 3.如何发送顺序消息

Rocketmq能够保证消息严格顺序,
但是Rocketmq需要producer保证顺序消息
按顺序发送到同一个queue中,
比如购买流程(1)下单(2)支付(3)支付成功,
这三个消息需要根据特定规则将这个
三个消息按顺序发送到一个queue
如何实现把顺序消息发送到同一个queue:
一般消息是通过轮询所有队列发送的,
顺序消息可以根据业务比如说订单号orderId
相同的消息发送到同一个队列,
或者同一用户userId发送到同一队列等等
messageQueueList [orderId%messageQueueList.size()]
messageQueueList [userId%messageQueueList.size()]

producer 4.如何发布分布式事务消息

先引入官方文档图:

分布式事物是基于二阶段提交的

  1. 一阶段,向broker发送一条prepared的消息,返回消息的offset即消息地址commitLog中消息偏移量。Prepared状态消息不被消费 发送消息ok,执行本地事物分支, 本地事物方法需要实现rocketmq的回调接口
  2. LocalTransactionExecuter, 处理本地事物逻辑返回处理的事物状态LocalTransactionState
  3. 二阶段,处理完本地事物中业务得到事物状态, 根据offset查找到commitLog中的prepared消息,设置消息状态commitType或者rollbackType, 让后将信息添加到commitLog中, 其实二阶段生成了两条消息

事物消息发送

producer 5.消息在落地broker落地之普通消息

Broker根据producer请求的RequestCode.SEND_MESSAGE选择对应的处理器SendMessageProcessor

根据请求消息内容构建消息内部结构MessageExtBrokerInner

调DefaultMessageStore加消息写入commitlog

producer 6.消息在落地broker落地之事务消息

1. 消息落地

commitLog针对事物消息的处理,消息的第20位开始的八位记录是的消息在逻辑队列中的queueoffset,但是针对事物消息为preparedType和rollbackType的存储的是事物状态表的索引偏移量

2. 分发事物消息

分发消息位置信息到ConsumeQueue: 事物状态为preparedType和rollbackType的消息不会将请求分发到ConsumeQueue中去,即不处理,所以不会被消息更新transactionstable table:如果是prepared消息记,通过TransactionStateService服务将消息加到存储事务状态的表格tranStateTable的文件中;如果是commitType和rollbackType消息, 修改事物状态表格tranStateTable中的消息状态。记录Transaction Redo Log日志: 记录了 commitLogOffset, msgSize,preapredTransactionOffset, storeTimestamp。

3. 事物状态表

事物状态表是有MapedFileQueue将多个文件组成一个连续的队列,它的存储单元是定长为24个字节的数据,tranStateTableOffset可以认为是事物状态消息的个数,索引偏移量, 它的值是 tranStateTable.getMaxOffset()/ TSStoreUnitSize

4. 事物回查

定时回查线程会定时扫描(默认每分钟)每个存储事务状态的表格文件,遍历存储事务状态的表格记录,如果是已经提交或者回滚的消息调过过,如果是prepared状态的如果消息小于事务回查至少间隔时间(默认是一分钟)跳出终止遍历调transactionCheckExecuter.gotocheck方法向producer回查事物状态,根据group随机选择一台producer查询消息,根据commitLogOffset和msgSize到commitlog查找消息向Producder发起请求,请求code类型为CHECK_TRANSACTION_STATE,producer的DefaultMQProducerImpl.checkTransactionState()方法来处理broker定时回调的请求,这里构建一个Runnable任务异步执行producer注册的回调接口,处理回调,在调endTransactionOneway向broker发送请求更新事物消息的最终状态无Prepared消息,且遍历完,则终止扫描这个文件的定时任务

5. 事物消息的load&recover

TransactionStateService.load ()事物状态服务加载, 加载只是建立文件映射redoLog队列恢复,加载本地redoLog文件tranStateTable事物状态表, 加载本地tranStateTable文件recover:

正常恢复

  • 利用tranRedoLog文件的recover
  • 利用tranStateTable文件重建事物状态表

异常恢复

先按照正常流程恢复TranRedo LogcommitLog异常恢复,commitLog根据checkpoint时间点重新生成 redolog,重新分发消息DispatchRequest,分发消息到位置信息到ConsumeQueue,更新Transaction State Table,记录TransactionRedo Log,删除事物状态表tranStateTable,通过RedoLog全量恢复StateTable,重头扫描RedoLog, 过滤出所有prepared状态的消息, 将commit或者rollback的消息对应的prepared消息删除,重建StateTable, 将上面过滤出的prepared消息,添加到事物状态表文件中

这个事物状态表transstable的作用是定期(1分钟)将状态为prepared事物回查producer端redolog这个队列其实标记消费到哪了,事物状态的恢复根本上是有commitlog来做的,consumer主要2中模式pull and push

写在最后:

  • 针对于Java程序员,笔者最近整理了一些面试真题,思维导图,程序人生等PDF学习资料;
  • 关注私信我"86",即可获取!
  • 希望读到这的您能点个小赞和关注下我,以后还会更新技术干货,谢谢您的支持!

相关推荐

辞旧迎新,新手使用Containerd时的几点须知

相信大家在2020年岁末都被Kubernetes即将抛弃Docker的消息刷屏了。事实上作为接替Docker运行时的Containerd在早在Kubernetes1.7时就能直接与Kubelet集成使...

分布式日志系统ELK+skywalking分布式链路完整搭建流程

开头在分布式系统中,日志跟踪是一件很令程序员头疼的问题,在遇到生产问题时,如果是多节点需要打开多节点服务器去跟踪问题,如果下游也是多节点且调用多个服务,那就更麻烦,再者,如果没有分布式链路,在生产日志...

Linux用户和用户组管理

1、用户账户概述-AAA介绍AAA指的是Authentication、Authorization、Accounting,即认证、授权和审计。?认证:验证用户是否可以获得权限,是3A的第一步,即验证身份...

linux查看最后N条日志

其实很简单,只需要用到tail这个命令tail-100catalina.out输入以上命令,就能列出catalina.out的最后100行。...

解决linux系统日志时间错误的问题

今天发现一台虚拟机下的系统日志:/var/log/messages,文件时间戳不对,跟正常时间差了12个小时。按网上说的执行了servicersyslogrestart重启syslog服务,还是不...

全程软件测试(六十二):软件测试工作如何运用Linux—读书笔记

从事过软件测试的小伙们就会明白会使用Linux是多么重要的一件事,工作时需要用到,面试时会被问到,简历中需要写到。对于软件测试人员来说,不需要你多么熟练使用Linux所有命令,也不需要你对Linux...

Linux运维之为Nginx添加错误日志(error_log)配置

Nginx错误日志信息介绍配置记录Nginx的错误信息是调试Nginx服务的重要手段,属于核心功能模块(nginx_core_module)的参数,该参数名字为error_log,可以放在不同的虚机主...

Linux使用swatchdog实时监控日志文件的变化

1.前言本教程主要讲解在Linux系统中如何使用swatchdog实时监控日志文件的变化。swatchdog(SimpleWATCHDOG)是一个简单的Perl脚本,用于监视类Unix系统(比如...

syslog服务详解

背景:需求来自于一个客户想将服务器的日志转发到自己的日志服务器上,所以希望我们能提供这个转发的功能,同时还要满足syslog协议。1什么是syslog服务1.1syslog标准协议如下图这里的fa...

linux日志文件的管理、备份及日志服务器的搭建

日志文件存放目录:/var/log[root@xinglog]#cd/var/log[root@xinglog]#lsmessages:系统日志secure:登录日志———————————...

运维之日志管理简介

日志简介在运维过程中,日志是必不可少的东西,通过日志可以快速发现问题所在。日志分类日志分类,对不同的日志进行不同维度的分析。操作系统日志操作系统是基础,应用都是在其之上;操作系统日志的分析,可以反馈出...

Apache Log4j 爆核弹级漏洞,Spring Boot 默认日志框架就能完美躲过

这两天沸沸扬扬的Log4j2漏洞门事件炒得热火朝天:突发!ApacheLog4j2报核弹级漏洞。。赶紧修复!!|Java技术栈|Java|SpringBoot|Spring...

Linux服务器存在大量log日志,如何快速定位错误?

来源:blog.csdn.net/nan1996jiang/articlep/details/109550303针对大量log日志快速定位错误地方tail/head简单命令使用:附加针对大量log日志...

Linux中查看日志文件的正确姿势,求你别tail走天下了!

作为一个后端开发工程师,在Linux中查看查看文件内容是基本操作了。尤其是通常要分析日志文件排查问题,那么我们应该如何正确打开日志文件呢?对于我这种小菜鸡来说,第一反应就是cat,tail,vi(或...

分享几款常用的付费日志系统,献给迷茫的你!

概述在前一篇文章中,我们分享了几款免费的日志服务器。他们各有各的特点,但是大家有不同的需求,有时免费的服务器不能满足大家的需要,下面推荐几款付费的日志服务器。1.Nagios日志服务器Nagio...

取消回复欢迎 发表评论: