RocketMQ Broker主从复制流程

Posted 2022-11-12 乐观男孩

目录

• 复制过程
• 连接建立
• Slave Broker上报复制offset
• Master Broker读取Slave Broker上报的offset
• Master Broker复制消息给Slave Broker
• Slave Broker读取新数据
• 总结

复制过程

Broker主从复制主要逻辑在HAService类的内部，主要涉及以下组件：
1、HAClient：在Slave Broker角色才使用，其主要功能包括连接Master Broker、向Master Broker上报需要复制的offset、处理Master Broker的响应。
2、AcceptSocketService：接收Slave Broker的复制连接请求、创建HAConnection实例并进行启动。
3、ReadSocketService：解析HAClient请求的内容；
4、WriteSocketService：负责将数据写给Slave Broker。

连接建立

Slave Broker在启动时，HAService内的HAClient会与Master Broker建立连接

通过connectMaster()建立连接

Master Broker的HAService内的AcceptSocketService将会接受新的连接请求，并把新的连接封装成HAConnection实例

至此，专门用于Master Broker与Slave Broker复制数据的专属链接就已建立完成。

Slave Broker上报复制offset

Slave Broker将定时或接收到新数据时，向Master Broker上报下次待复制的offset。

reportSlaveMaxOffsetPlus()方法在接收到新数据时，其内部通过调用reportSlaveMaxOffset()方法上报offset

reportSlaveMaxOffset()方法执行上报offset具体实现

Master Broker读取Slave Broker上报的offset

Master Broker读取Slave Broker上报的offset，是通过HAConnection内的ReadSocketService实现的

通过processReadEvent()方法解析Slave Broker上报的offset

Master Broker复制消息给Slave Broker

当Slave Broker第一次上报offset的时候，上报的offset将会保存到slaveRequestOffset属性中，slaveRequestOffset属性值有以下三种情况：
（1）、slaveRequestOffset=-1：初始值，证明Slave Broker还未上报offset，此时不会执行复制操作
（2）、slaveRequestOffset=0：Slave Broker上报的offset为0，证明Slave Broker从未复制过数据，此时slaveRequestOffset的值为Master Broker当前最大offset - (102410241024)，即往前找1GB数据的offset开始复制
（3）、slaveRequestOffset>0：按照Slave Broker上报的offset开始同步

当slaveRequestOffset≥0之后，Master Broker就可以开始复制工作。整个复制工作分以下几个步骤：

1、根据offset读取消息：

（1）、根据offset查找MapperFile

（2）、根据offset读取内容

说明：根据上一篇文章《RocketMQ Broker处理新消息流程》所提及的，MapperFile里mappedByteBuffer的position和limit一直是不变的（精妙之处，可以做很多灵活处理），所以mappedByteBuffer.slice()能拿到整个Buffer。所以读取数据示意图如下（最终红色部分就是需要复制的数据）：

2、根据offset读取完内容后，将构建头部信息，头部信息主要包括即将复制数据的offset和数据大小

3、通过transferData()方法进行数据复制
（1）、先将头部信息写给Slave Broker，告知Slave Broker即将复制的offset和数据大小

（2）、接着复制具体的数据给Slave Broker

至此，如果复制工作正常执行，数据就会复制到Slave Broker。

Slave Broker读取新数据

Slave Broker读取新数据是在HAClient.processReadEvent()方法进行处理

Slave Broker读取新数据后，会解析头部信息和获取新的数据。新的数据会拼接到Slave Broker的CommitLog中，然后会按照先前介绍的步骤，将新消息写到MapperFile，然后再刷新数据到磁盘，完成数据复制的整个工作。

总结

1、RocketMQ的主从复制，实现方法比较常规，主要包括连接的建立、offset的上报，数据的复制等过程。
2、根据offset读取待复制内容的过程是精妙所在，也是难以理解的地方。其主要思想是保证原Buffer的positon和limit不动，然后通过slice操作生成新的Buffer，在新的Buffer上通过移动position后，再获取想要的数据。