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实现的。

RocketMQ源码-主从同步复制和异步复制