1 概述
2 相关类介绍
3 同步复制原理
4 异步复制原理
5 注意事项
1 概述
为了提供系统的可靠性,RocketMQ采用了主从复制机制,对于每个主Broker,可配置多个从Broker,主Broker接收生产者消息保存后,可通过同步或者异步的方式复制到从Broker,以此实现高可用。
采用同步复制,可以最大程度保证消息的可靠性,但是每次写消息都必须等待至少该消息被同步至一个从Broker中,因此也会影响系统吞吐量。异步复制则和同步复制正好相反,主Broker写完消息之后立刻返回,不用管该消息是否已经被复制到从Broker,因此吞吐量会高一点,但是如果在消息被复制到从Broker之前主Broker发生故障,那么有可能会造成未来得及复制到从Broker的信息丢失。
本文会简单介绍下RocketMQ主从同步的实现。
2 相关类介绍
HAService
是主从复制服务的主要实现类,通过内部相关组件实现接受从Broker连接请求、记录从Broker上报的复制进度等功能。
AcceptSocketService
主要负责接受从Broker的连接请求,接受到的每个从Broker的连接之后会新建HAConnection
对象实例。
HAConnection
主Broker管理的一系列从Broker连接,内部持有一个ReadSocketService
对象实例,负责接收从Broker定时上报的自己当前复制进度,也持有一个WriteSocketService
负责在通道可写时向从Broker发送需要复制的数据,完成主从复制。
HAClient
一个ServiceThread
实现类,如果是从Broker,则会尝试向主Broker建立连接,并定时向主Broker汇报自己的复制进度,之后监听OP_READ
事件(可参考笔者文章NIO SelectionKey事件理解),处理主Broker通过HAConnection.WriteSocketService
发送过来的需要复制的数据。
GroupTransferService
主从同步复制的实现类,如果是主从同步复制,则会向该类提交一个复制任务请求,并进入阻塞等待状态,该任务主要封装了当前主Broker的消息写进度,GroupTransferService
也是一个ServiceThread
,会定时获取HAConnection.ReadSocketService
接受到的从Broker的最大复制进度,然后对比所有的复制任务请求,如果已经从Broker最大复制进度已经大于请求内的进度要求,则唤醒该同步复制阻塞,阻塞和唤醒是通过CountDownLatch
实现的。
3 同步复制原理
BrokerController
通过DefaultMessageStore.putMessage
存储消息,后者则通过调用CommitLog.putMessage
进行实际的消息存储处理,CommitLog.putMessage
在将消息写入内存缓冲之后会先调用handleDiskFlush
进行同步或异步刷盘,之后会调用handleHA
进行主从复制处理。
handlerHA
方法定义如下:
//CommitLog
public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
//住过服务器状态为同步主Broker,则需要向HAService的
//GroupTransferService提交复制等待请求
if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
HAService service = this.defaultMessageStore.getHaService();
if (messageExt.isWaitStoreMsgOK()) {
// Determine whether to wait
//isSlaveOK则判断当前是否有从Broker连接,如果没有返回
//失败,所以如果集群只有一个主Broker,没有从Broker,但是
//配置该主Broker为同步模式,则会一直报
//SLAVE_NOT_AVAILABLE错误
//同时如果当前从Broker复制的最大进度离当前写入的位置偏差
//大于配置的指定值,则也会报错,这个判断就是为了保证
//从Broker不会落后主Broker太多
if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
//等待请求的复制进度为当前写开始位置+写入字节数
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
service.getWaitNotifyObject().wakeupAll();
//等待复制到该进度完成
boolean flushOK =
request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
if (!flushOK) {
log.error("do sync transfer other node, wait return, but failed, topic: " + messageExt.getTopic() + " tags: "
+ messageExt.getTags() + " client address: " + messageExt.getBornHostNameString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
}
}
// Slave problem
else {
// Tell the producer, slave not available
putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
}
}
}
}
GroupTransferService
接受了复制等待请求任务之后会在run
方法中定时查看从Broker上报的最大复制进度,会唤醒那些等待进度已达到的等待任务。
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
//定时调用doWaitTransfer方法
this.waitForRunning(10);
this.doWaitTransfer();
} catch (Exception e) {
log.warn(this.getServiceName() + " service has exception. ", e);
}
}
log.info(this.getServiceName() + " service end");
}
private void doWaitTransfer() {
synchronized (this.requestsRead) {
if (!this.requestsRead.isEmpty()) {
//对于提交上来的每个等待复制请求,判断是否达到
//预期的复制进度
for (CommitLog.GroupCommitRequest req : this.requestsRead) {
//如果从Broker上报的赋值进度大于请求期望的偏移
//则表示传输完成,也就是等待完成
boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
//否则的话,再尝试五次判断,每次间隔一秒
for (int i = 0; !transferOK && i < 5; i++) {
this.notifyTransferObject.waitForRunning(1000);
transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
}
//如果五次依然失败,则记录日志
if (!transferOK) {
log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
}
//令等待同步复制的handleHA方法返回,并通过transferOK
//告诉其等待结果
req.wakeupCustomer(transferOK);
}
this.requestsRead.clear();
}
}
}
4 异步复制原理
其实异步复制原理比较简单,如果Broker配置为异步复制,则在CommitLog.putMessage
写入消息之后,调用handleHA
方法不会做任何操作,写入之后根本不用管从Broker的复制进度,复制完全是由后台HAConnection.WriteSocketService
服务在监听到有从Broker连接可写时,向其写待复制的数据。每个从Broker发送进度则由从Broker定时汇报的自身当前已复制进度控制,该汇报由HAConnection.ReadSocketService
负责处理,从Broker汇报上来的最大复制进度则用于第3节介绍的同步复制的等待复制任务阻塞的线程。
5 注意事项
RocketMQ各组件通信比如Broker和Namesrv、Producer和Namesrv、Consumer和Namesrv、Producer以及Consumer和Broker之间的通信都是基于Netty实现的,但是本文介绍的主从复制实现中的网络交互都是基于原生Java NIO实现的。
网友评论