一、主从同步工作原理

ReadSocketService线程读取从Broker复制进度

三、读写分离机制

为了提高消息消费的高可用性避免Broker发生单点故障引起存储在Broker上的消息无法及时消费

RocketMQ引入Broker主备机制即消息消费到达主服务器后需要将消息同步到消息从服务器如果主服务器Broker宕机后消息消费者可以从从服务器拉取消息。

下图所示是Broker的HA交互机制流程图及类图。

主从同步模式分为同步、异步。

step1主服务器启动并在特定端口上监听从服务器的连接step2从服务器主动连接主服务器主服务器接收客户端的连接并建立相关TCP连接step3从服务器主动向主服务器发送待拉取消息偏移量主服务器解析请求并返回消息给从服务器step4从服务器保存消息并继续发送新的消息同步请求。

1.

主从配置

参考rocketmq-distribution项目的conf目录下有2主2从异步HA配置2m-2s-async、2主2从同步HA配置2m-2s-sync。

以下1主1从异步HA配置实例如下。

主Broker配置

truenamesrvAddr192.168.1.55:9876;172.17.0.3:9876

从Broker配置

0时则为从0时则为主brokerRole角色为SLAVE从刷盘类型为ASYNC_FLUSH异步刷盘。

brokerClusterName

truenamesrvAddr192.168.1.55:9876;172.17.0.3:9876

启动HA

org.apache.rocketmq.store.DefaultMessageStore#start是Broker启动方法如下图所示是其调用链及相关HA部分代码。

/***

BrokerController#startBasicService()*

throws

(!messageStoreConfig.isEnableDLegerCommitLog()

!this.messageStoreConfig.isDuplicationEnable())

{this.haService.init(this);}......if

(this.haService

{this.haService.start();}......

org.apache.rocketmq.store.ha.DefaultHAService#init是HAService初始化方法如下代码所示。

注意从Broker的broker.conf配置的brokerRole为SLAVE才能创建HAClient从Broker注册到主Broker。

Override

defaultMessageStore;this.acceptSocketService

new

DefaultAcceptSocketService(defaultMessageStore.getMessageStoreConfig());this.groupTransferService

new

(this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()

BrokerRole.SLAVE)

DefaultHAClient(this.defaultMessageStore);}this.haConnectionStateNotificationService

new

HAConnectionStateNotificationService(this,

org.apache.rocketmq.store.ha.DefaultHAService#start是HAService启动方法。

注意

org.apache.rocketmq.store.ha.DefaultHAService.AcceptSocketService主Broker接收从Broker的连接事件org.apache.rocketmq.store.ha.GroupTransferService负责主Broker向从Broker发送同步数据org.apache.rocketmq.store.ha.HAClient从Broker向主Broker发送连接事件Broker启动时根据配置brokerRole配置ASYNC_MASTER、SYNC_MASTER、SLAVE判定Broker是主还是从。

若是Slave角色在broker配置文件中获取haMasterAddress并更新至masterAddress但是haMasterAddress配置为空则启动成功但是不会执行HA。

/***

AcceptSocketService}接收从Broker的注册事件方法是{link

AcceptSocketService#beginAccept()}*

step2启动{link

AcceptSocketService}线程监听从Broker发送心跳*

step3同步数据{link

GroupTransferService}线程启动主Broker向从Broker发送数据*

step4启动从Broker{link

主接收从Broker的连接事件SelectionKey.OP_ACCEPT连接事件this.acceptSocketService.beginAccept();//

启动主Broker线程this.acceptSocketService.start();//

主Broker同步数据线程启动this.groupTransferService.start();this.haConnectionStateNotificationService.start();//

启动从Broker{link

org.apache.rocketmq.store.ha.DefaultHAClient是从Broker向主Broker的发送连接事件的核心类是个线程。

其主要属性如下代码所示。

Socket读缓存区大小4M

反馈HA的复制进度从Broker的Commitlog文件的最大偏移量

private

ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

private

ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

private

org.apache.rocketmq.store.ha.DefaultHAClient#run是HAClient启动执行任务其调用链和代码如下。

DefaultHAClient#connectMaster()从Broker连接到主Broker。

DefaultHAClient#transferFromMaster()向主发送HA进度处理从主Broker发送过来的消息并commit所有消息追加到Commitlog文件内存映射缓存中。

/***

DefaultHAClient#connectMaster()}从Broker连接到主Broker*

{link

DefaultHAClient#transferFromMaster()}向主发送HA进度处理从主Broker发送过来的消息并commit所有消息追加到Commitlog文件内存映射缓存中*/

Override

{log.info(this.getServiceName()

service

started);this.flowMonitor.start();while

(!this.isStopped())

this.masterHaAddress.get());this.waitForRunning(1000

5);}continue;case

向主发送HA进度处理从主Broker发送过来的消息并commit所有消息追加到Commitlog文件内存映射缓存中if

{//

没有可拉取消息时设置READY状态closeMasterAndWait();continue;}break;default:this.waitForRunning(1000

2);continue;}long

this.defaultMessageStore.getMessageStoreConfig().getHaHousekeepingInterval())

housekeeping,

interval);this.closeMaster();log.warn(AutoRecoverHAClient,

master

{log.warn(this.getServiceName()

service

e);this.closeMasterAndWait();}}log.info(this.getServiceName()

service

注意一旦HAClient线程启动后在状态READY、TRANSFER来回变化READY状态下发送从Broker连接事件到主Broker开启Socket连接TRANSFER状态下主从发送相关数据信息如从向主发送HA复制进度currentReportedOffset即从Broker的Commitlog文件的最大偏移量主向从发送同步消息。

org.apache.rocketmq.store.ha.DefaultHAClient#connectMaster是从Broker连接到主Broker的核心方法其代码如下。

/***

Broker启动时若是Slave角色从broker配置文件中获取haMasterAddress并更新至masterAddress*

若是Slave角色但是haMasterAddress配置为空则启动成功但是不会执行HA*

return

根据地址创建SocketAddress对象SocketAddress

socketAddress

RemotingUtil.string2SocketAddress(addr);//

获取SocketChannelthis.socketChannel

RemotingUtil.connect(socketAddress);if

(this.socketChannel

SocketChannel注册OP_READ网络读事件this.socketChannel.register(this.selector,

SelectionKey.OP_READ);log.info(HAClient

connect

addr);this.changeCurrentState(HAConnectionState.TRANSFER);}}//

获取Commitlog最大偏移量HA同步进度this.currentReportedOffset

this.defaultMessageStore.getMaxPhyOffset();this.lastReadTimestamp

System.currentTimeMillis();}return

this.socketChannel

org.apache.rocketmq.store.ha.DefaultHAService.AcceptSocketService是主Broker接收从Broker连接事件的实现类是一个线程。

其主要属性如下代码所示。

端口号

org.apache.rocketmq.store.ha.DefaultHAService.AcceptSocketService#beginAccept方法定义了主Broker监听从Broker的连接事件。

/***

ServerSocketChannel.open();this.selector

RemotingUtil.openSelector();this.serverSocketChannel.socket().setReuseAddress(true);

TCP可重复使用this.serverSocketChannel.socket().bind(this.socketAddressListen);

绑定监听端口if

messageStoreConfig.getHaListenPort())

{messageStoreConfig.setHaListenPort(this.serverSocketChannel.socket().getLocalPort());log.info(OS

picked

messageStoreConfig.getHaListenPort());}this.serverSocketChannel.configureBlocking(false);

非阻塞模式this.serverSocketChannel.register(this.selector,

注册OP_ACCEPT连接事件

org.apache.rocketmq.store.ha.DefaultHAService.AcceptSocketService#run监听到从Broker连接事件的任务处理为每个连接事件创建org.apache.rocketmq.store.ha.HAConnection对象并启动负责M-S的数据同步逻辑。

/***

{log.info(this.getServiceName()

service

选择器每1s处理一次连接就绪事件this.selector.select(1000);SetSelectionKey

selected

this.selector.selectedKeys();if

(selected

若是连接事件时创建SocketChannelSocketChannel

(sc

{DefaultHAService.log.info(HAService

receive

sc.socket().getRemoteSocketAddress());try

{//

每一个连接创建HAConnection并启动负责M-S的数据同步逻辑HAConnection

conn

createConnection(sc);conn.start();DefaultHAService.this.addConnection(conn);}

catch

k.readyOps());}}selected.clear();}}

catch

{log.error(this.getServiceName()

service

e);}}log.info(this.getServiceName()

service

org.apache.rocketmq.store.ha.DefaultHAConnection创建并启动时启动读、写线程服务。

其关键属性如下代码所示。

private

writeSocketService主Broker向从Broker写数据服务类private

ReadSocketService

readSocketService主Broker读取从Broker数据服务类

private

org.apache.rocketmq.store.ha.DefaultHAClient#transferFromMaster是从Broker与主Broker传输数据的核心方法代码如下所示该方法有两大功能

从Broker向主Broker反馈HA复制进度即currentReportedOffset从Broker的Commitlog文件的最大偏移量方法org.apache.rocketmq.store.ha.DefaultHAClient#reportSlaveMaxOffset执行。

从Broker接收主BrokerHA同步消息内容方法org.apache.rocketmq.store.ha.DefaultHAClient#processReadEvent执行。

/***

向主反馈HA复制进度处理从主Broker发送过来的消息并commit所有消息追加到Commitlog文件内存映射缓存中*

{link

DefaultHAClient#reportSlaveMaxOffset(long)}向主反馈HA复制进度即currentReportedOffset从Broker的Commitlog文件的最大偏移量*

{link

DefaultHAClient#processReadEvent()}处理从主Broker发送过来的消息并commit所有消息追加到Commitlog文件内存映射缓存中*/

private

this.reportSlaveMaxOffset(this.currentReportedOffset);if

(!result)

false;}}this.selector.select(1000);//

(!result)

org.apache.rocketmq.store.ha.DefaultHAClient#reportSlaveMaxOffset向主Broker反馈HA复制进度代码如下。

/***

对于Master端本次请求拉取的偏移量也可以理解为同步ACK*

Buffer#hasRemaining()}判断缓存内容是否完全写入SocketChannel基于NIO模式的写范例*

param

偏移量写入ByteBufferthis.reportOffset.position(0);

写缓存位置this.reportOffset.limit(8);

写缓存字节长度this.reportOffset.putLong(maxOffset);

偏移量写入ByteBuffer//

读模式this.reportOffset.position(0);this.reportOffset.limit(8);//

循环并判定ByteBuffer是否完全写入SocketChannelfor

(int

this.reportOffset.hasRemaining();

{try

{this.socketChannel.write(this.reportOffset);}

catch

{log.error(this.getServiceName()

reportSlaveMaxOffset

this.defaultMessageStore.getSystemClock().now();return

!this.reportOffset.hasRemaining();

ReadSocketService线程读取从Broker复制进度

org.apache.rocketmq.store.ha.DefaultHAConnection.ReadSocketService#processReadEvent是主Broker读取从Broker拉取消息的请求获取内容是HA复制进度。

其代码如下看出主Broker获取从Broker的HA复制进度后赋值给DefaultHAConnection#slaveRequestOffset属性后立即唤醒GroupTransferService线程执行消息同步。

/***

step1判定byteBufferRead是否有剩余空间没有则{link

Buffer#flip()}*

step2用剩余空间从SocketChannel读数据到缓存中读取到的内容是从Broker拉取消息的偏移量*

return*/

0;/*byteBufferRead没有剩余空间时则position

limit

(!this.byteBufferRead.hasRemaining())

ByteBuffer重置处理this.processPosition

0;}//

ByteBuffer有剩余空间循环至byteBufferRead没有剩余空间while

(this.byteBufferRead.hasRemaining())

{try

this.socketChannel.read(this.byteBufferRead);if

(readSize

DefaultHAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();//

读取内容长度

((this.byteBufferRead.position()

{int

(this.byteBufferRead.position()

8);long

this.byteBufferRead.getLong(pos

pos;//

从Broker反馈已完成的偏移量DefaultHAConnection.this.slaveAckOffset

readOffset;//

(DefaultHAConnection.this.slaveRequestOffset

{DefaultHAConnection.this.slaveRequestOffset

DefaultHAConnection.this.clientAddress

request

通知等待同步HA复制结果的发送消息线程DefaultHAConnection.this.haService.notifyTransferSome(DefaultHAConnection.this.slaveAckOffset);}}

else

DefaultHAConnection.this.clientAddress

0);return

ReadSocketService线程读取从Broker发送的HA复制进度由org.apache.rocketmq.store.ha.DefaultHAConnection.WriteSocketService根据DefaultHAConnection#slaveRequestOffset获取主Broker还没有同步的所有消息进行HA同步。

其如下代码所示WriteSocketService#run方法是同步消息核心逻辑。

/***

step1slaveRequestOffset为-1时说明主Broker没有收到从Broker的拉取请求忽略本次写事件*

step2nextTransferFromWhere为-1时说明初次传输计算nextTransferFromWhere待传输offset*

上次传输完当前时间

step4根据从Broker待拉取消息offset查找之后的所有可读消息*

step5待同步消息总大小

{log.info(this.getServiceName()

service

slaveRequestOffset为-1时说明主Broker没有收到从Broker的拉取请求忽略本次写事件if

(-1

DefaultHAConnection.this.slaveRequestOffset)

{Thread.sleep(10);continue;}/*nextTransferFromWhere为-1时说明初次传输初次传输时计算nextTransferFromWhere待传输offset*///

初次传输if

DefaultHAConnection.this.slaveRequestOffset)

{long

DefaultHAConnection.this.haService.getDefaultMessageStore().getCommitLog().getMaxOffset();masterOffset

masterOffset-

DefaultHAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getMappedFileSizeCommitLog());if

(masterOffset

DefaultHAConnection.this.slaveRequestOffset;}log.info(master

transfer

DefaultHAConnection.this.clientAddress

and

DefaultHAConnection.this.slaveRequestOffset);}/*判断上次是否传输完上次传输完当前时间

发送心跳包时间间隔发送心跳包长度:

size消息长度0避免长连接由空闲而关闭上次传输没有完成继续传输忽略本次写事件*///

(this.lastWriteOver)

DefaultHAConnection.this.haService.getDefaultMessageStore().getSystemClock().now()

时间间隔

DefaultHAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaSendHeartbeatInterval())

{//

size消息长度0避免长连接由空闲而关闭this.byteBufferHeader.position(0);this.byteBufferHeader.limit(headerSize);this.byteBufferHeader.putLong(this.nextTransferFromWhere);this.byteBufferHeader.putInt(0);this.byteBufferHeader.flip();//

(!this.lastWriteOver)continue;}}//

lastWriteOver为false则上次传输没有完成则继续传输else

{//

继续传输上次拉取请求还未完成则忽略本次写事件this.lastWriteOver

(!this.lastWriteOver)continue;}//

根据从Broker待拉取消息offset查找之后的所有可读消息SelectMappedBufferResult

selectResult

DefaultHAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);if

(selectResult

DefaultHAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize())

{size

DefaultHAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();}int

canTransferMaxBytes

flowMonitor.canTransferMaxByteNum();if

(size

flowMonitor.maxTransferByteInSecond()

flowMonitor.getTransferredByteInSecond()

System.currentTimeMillis();}size

thisOffset

this.nextTransferFromWhere;this.nextTransferFromWhere

size;

下一次写入的offsetselectResult.getByteBuffer().limit(size);this.selectMappedBufferResult

selectResult;//

传输size大小的消息内容不一定是完整的消息this.byteBufferHeader.position(0);this.byteBufferHeader.limit(headerSize);this.byteBufferHeader.putLong(thisOffset);this.byteBufferHeader.putInt(size);this.byteBufferHeader.flip();//

else

{DefaultHAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);}}

catch

{DefaultHAConnection.log.error(this.getServiceName()

service

e);break;}}DefaultHAConnection.this.haService.getWaitNotifyObject().removeFromWaitingThreadTable();if

null)

{this.selectMappedBufferResult.release();}changeCurrentState(HAConnectionState.SHUTDOWN);this.makeStop();readSocketService.makeStop();haService.removeConnection(DefaultHAConnection.this);SelectionKey

this.socketChannel.keyFor(this.selector);if

(sk

{this.selector.close();this.socketChannel.close();}

catch

{DefaultHAConnection.log.error(,

e);}DefaultHAConnection.log.info(this.getServiceName()

service

org.apache.rocketmq.store.ha.GroupTransferService该类负责将主从同步复制结束后通知阻塞的消息发送者线程。

同步主从Broker模式即消息刷磁盘后继续等待新消息被传输到从Broker等待传输结果并通知消息发送线程。

org.apache.rocketmq.store.CommitLog#asyncPutMessage是消息生产者发送消息到Broker时执行存储消息参考《RocketMQ5.0.0消息存储二_消息存储流程》该方法会根据同步或异步模式默认来执行org.apache.rocketmq.store.CommitLog#handleDiskFlushAndHA方法完成刷盘和HA复制方法调用链如下。

生产者把消息发送到Broker完成commit操作消息提交到文件内存映射中

随后根据同步/异步模式完成刷盘和HA。

HA操作时把消息提交请求添加到org.apache.rocketmq.store.ha.GroupTransferService.requestsWrite是主Broker待需要HA的的集合。

以下是org.apache.rocketmq.store.CommitLog#handleHA的代码。

private

CompletableFuturePutMessageStatus

result,

CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);}HAService

haService

this.defaultMessageStore.getHaService();long

nextOffset

this.defaultMessageStore.getMessageStoreConfig().getSlaveTimeout(),

needAckNums);haService.putRequest(request);haService.getWaitNotifyObject().wakeupAll();return

request.future();

HAService启动时会启动GroupTransferService线程。

GroupTransferService#run执行任务如下代码所示。

Override

{log.info(this.getServiceName()

service

间隔10sthis.waitForRunning(10);//

主从同步复制结束后通知阻塞的消息发送者线程this.doWaitTransfer();}

catch

{log.warn(this.getServiceName()

service

e);}}log.info(this.getServiceName()

service

其中执行waitForRunning()方法时会去执行org.apache.rocketmq.store.ha.GroupTransferService#swapRequests方法使得requestsWrite与requestsRead两个集合对调

private

ListCommitLog.GroupCommitRequest

requestsWrite主Broker待需要HA的消息集合private

volatile

ListCommitLog.GroupCommitRequest

private

{ListCommitLog.GroupCommitRequest

tmp

this.requestsWrite;this.requestsWrite

this.requestsRead;this.requestsRead

tmp;

org.apache.rocketmq.store.ha.GroupTransferService#doWaitTransfer方法是主从同步复制结束后通知阻塞的消息发送者线程如下代码所示。

/***

step1遍历消息提交请求内存提交到Commitlog文件的内存映射*

private

commit请求即内存提交到Commitlog文件的内存映射for

req

MixAll.ALL_ACK_IN_SYNC_STATE_SET;for

(int

等待1sthis.notifyTransferObject.waitForRunning(1000);}if

req.getAckNums()

消息生产者发送消息后返回下一条消息的偏移量transferOK

haService.getPush2SlaveMaxOffset().get()

req.getNextOffset();continue;}if

this.haService

autoSwitchHAService.getSyncStateSet();if

(syncStateSet.size()

(syncStateSet.contains(autoSwitchHAConnection.getSlaveAddress())

autoSwitchHAConnection.getSlaveAckOffset()

req.getNextOffset())

从完成复制后唤醒消息发送者线程req.wakeupCustomer(transferOK

PutMessageStatus.FLUSH_SLAVE_TIMEOUT);}this.requestsRead.clear();}}

三、读写分离机制

RocketMQ读写分离与其他中间件的实现方式完全不同RocketMQ是消费者首先向主服务器发起拉取消息请求然后主服务器返回一批消息然后会根据主服务器负载压力与主从同步情况向从服务器建议下次消息拉取是从主服务器还是从从服务器拉取。

RocketMQ根据MessageQueu查找Broker地址的唯一依据是brokerName。

Broker组织中根据brokerName获取一组Broker服务器M-S它们的brokerName相同但brokerId不同主服务器的brokerId为0从服务器的brokerId大于0。

其方法是org.apache.rocketmq.client.impl.factory.MQClientInstance#findBrokerAddressInSubscribe

详细消费拉取消息时实现读写分离机制见后续章节参考《RocketMQ5.0.0消息消费一

四、参考资料

【RocketMQ】学习RocketMQ必须要知道的主从同步原理_午睡的猫…的博客-CSDN博客_rocketmq主从同步原理

【RocketMQ】主从同步实现原理

RocketMQ5.0.0消息存储二_消息存储流程_爱我所爱0505的博客-CSDN博客_rocketmq

消息写入流程RocketMQ5.0.0消息存储三_消息转发与恢复机制_爱我所爱0505的博客-CSDN博客

RocketMQ5.0.0消息存储四_刷盘机制_爱我所爱0505的博客-CSDN博客