RocketMQ源码解析-主从同步原理(HA)

1、关键组件

主从同步的实现逻辑主要在HAService中，在它的构造函数中实例化了几个对象同时在start()方法内执行启动：

public class HAService {
    public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
        this.defaultMessageStore = defaultMessageStore;
        this.acceptSocketService =
            new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
        this.groupTransferService = new GroupTransferService();
        this.haClient = new HAClient();
    }

    ......

    public void start() throws Exception {
        this.acceptSocketService.beginAccept();
        this.acceptSocketService.start();
        this.groupTransferService.start();
        this.haClient.start();
    }
}

首先了解一下HAService的构造函数中的内容究竟是干什么的：

AcceptSocketService：主要是处理从节点的连接，调用AcceptSocketService#beginAccept()方法，这一步主要是进行端口绑定，在端口上监听从节点的连接请求；调用AcceptSocketService#start()方法启动服务，这一步主要为了处理从节点的连接请求，与从节点建立连接（可以看做是运行在master节点的）。
GroupTransferService:主要用于在主从同步的时候，等待数据传输完毕（可以看做是运行在master节点的。
HAClient：里面与master节点建立连接，向master汇报主从同步进度并存储master发送过来的同步数据（可以看做是运行在slave从节点的）。

了解完HAService中的组件，而且看到在start()方法中启动了各个组件，那么HAService在何时被启动的呢？

还记得之前在记录broker时，看过BrokerController#initialize()初始化方法内，同时也构建了DefaultMessageStore对象，它作为HAService构造函数的入参，定义的start()方法中就包含HAService的启动

1).构建DefaultMessageStore以及start()启动

//BrokerController.class
public class BrokerController {
    private MessageStore messageStore;
    //broekr初始化
    public boolean initialize() throws CloneNotSupportedException {
         .......
         this.messageStore =  new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,this.brokerConfig);
         .......
    }
    //borker启动
    public void start() throws Exception {
        if (this.messageStore != null) {
            this.messageStore.start();
        }
      }
    }
}

2)实例化HAServer以及start()启动

//DefaultMessageStore.class
public class DefaultMessageStore implements MessageStore {
    private final HAService haService;
    ......

    public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
        ......
        //实例化HAService
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {
            this.haService = new HAService(this);
        } else {
            this.haService = null;
        }
        ......
    }

public void start() throws Exception {
        ......
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {
            //启动HA
            this.haService.start();
            this.handleScheduleMessageService(messageStoreConfig.getBrokerRole());
        }
        ......
    }
}

2.主从同步流程

2.1.绑定端口，监听连接请求

AcceptSocketService#beginAccept方法里面首先获取了ServerSocketChannel，然后进行端口绑定，并在selector上面注册了OP_ACCEPT事件的监听，监听从节点的连接请求：

class AcceptSocketService extends ServiceThread {
        /**
         * 监听从节点的连接
         *
         * @throws Exception If fails.
         */
        public void beginAccept() throws Exception {
            // 创建ServerSocketChannel
            this.serverSocketChannel = ServerSocketChannel.open();
            // 获取selector
            this.selector = RemotingUtil.openSelector();
            this.serverSocketChannel.socket().setReuseAddress(true);
            // 绑定端口：10912
            this.serverSocketChannel.socket().bind(this.socketAddressListen);
            // 设置非阻塞
            this.serverSocketChannel.configureBlocking(false);
            // 注册OP_ACCEPT连接事件的监听
            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        }
}

2.2master节点处理连接

因为继承了ServiceThread，所以被调用start()启动方法后，会另外开启一个线程执行run()代码，这块就是处理连接请求：

public class HAService {
    class AcceptSocketService extends ServiceThread {
        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");
            // 如果服务未停止
            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);
                    // 获取监听到的事件
                    Set<SelectionKey> selected = this.selector.selectedKeys();
                    // 处理事件
                    if (selected != null) {
                        for (SelectionKey k : selected) {
                            // 如果是连接事件
                            if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                                SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
                                if (sc != null) {
                                    HAService.log.info("HAService receive new connection, "
                                        + sc.socket().getRemoteSocketAddress());
                                    try {
                                        // 创建HAConnection，建立连接
                                        HAConnection conn = new HAConnection(HAService.this, sc);
                                        // 启动
                                        conn.start();
                                        //添加连接
                                        HAService.this.addConnection(conn);
                                    }
                                    ...
                       
        }
    }
}

从selector中获取到监听到的事件；
如果是OP_ACCEPT连接事件，创建与从节点的连接对象HAConnection，与从节点建立连接，然后调用HAConnection的start方法进行启动，并创建的HAConnection对象加入到连接集合中，HAConnection中封装了Master节点和从节点的数据同步逻辑；

2.3HAClient

HAClient同样也继承了ServiceThread

public void run() {
            log.info(this.getServiceName() + " service started");
            //是否执行
            while (!this.isStopped()) {
                try {
                    //连接Master
                    if (this.connectMaster()) {
                        //判断时间间隔是否合法
                        if (this.isTimeToReportOffset()) {
                            // 发送同步偏移量,传入的参数是当前的主从复制偏移量currentReportedOffset
                            boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                            //返回不对则关闭连接
                            if (!result) {
                                this.closeMaster();
                            }
                        }

                        ......
}
        }

2.3.1slave与主节点建立连接

connectMaster()方法执行连接主节点操作

 class HAClient extends ServiceThread {
    // 当前的主从复制进度
    private long currentReportedOffset = 0;

        private boolean connectMaster() throws ClosedChannelException {
            if (null == socketChannel) {
                String addr = this.masterAddress.get();
                if (addr != null) {
                    // 将地址转为SocketAddress
                    SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
                    if (socketAddress != null) {
                        // 连接master
                        this.socketChannel = RemotingUtil.connect(socketAddress);
                        if (this.socketChannel != null) {
                            // 注册OP_READ可读事件监听
                            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
                        }
                    }
                }
                // 获取CommitLog中当前最大的偏移量
                this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                // 更新上次写入时间
                this.lastWriteTimestamp = System.currentTimeMillis();
            }
            return this.socketChannel != null;
        }
}

2.3.2处理网络可读事件

processReadEvent()方法中处理了可读事件，也就是处理Master节点发送的同步数据，首先从socketChannel中读取数据到byteBufferRead中，byteBufferRead是读缓冲区，读取数据的方法会返回读取到的字节数，对字节数大小进行判断：

  class HAClient extends ServiceThread {
        // 读缓冲区，会将从socketChannel读入缓冲区
        private ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

        private boolean processReadEvent() {
            int readSizeZeroTimes = 0;
            while (this.byteBufferRead.hasRemaining()) {
                try {
                    // 从socketChannel中读取数据到byteBufferRead中，返回读取到的字节数
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    if (readSize > 0) {
                        // 重置readSizeZeroTimes
                        readSizeZeroTimes = 0;
                        // 处理数据
                        boolean result = this.dispatchReadRequest();
                        if (!result) {
                            log.error("HAClient, dispatchReadRequest error");
                            return false;
                        }
                    } else if (readSize == 0) {
                        // 记录读取到空数据的次数
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        log.info("HAClient, processReadEvent read socket < 0");
                        return false;
                    }
                } catch (IOException e) {
                    log.info("HAClient, processReadEvent read socket exception", e);
                    return false;
                }
            }

            return true;
        }
  }

如果可读字节数大于0表示有数据需要处理，调用dispatchReadRequest方法进行处理;
如果可读字节数为0表示没有可读数据，此时记录读取到空数据的次数，如果连续读到空数据的次数大于3次，将终止本次处理;

2.3.3消息写入ComitLog

dispatchReadRequest方法中会将从节点读取到的数据写入CommitLog，dispatchPosition记录了已经处理的数据在读缓冲区中的位置，从读缓冲区byteBufferRead获取剩余可读取的字节数，如果可读数据的字节数大于一个消息头的字节数（12个字节），表示有数据还未处理完毕，反之表示消息已经处理完毕结束处理。

 private boolean dispatchReadRequest() {
            // 消息头大小
            final int msgHeaderSize = 8 + 4; // phyoffset + size
            int readSocketPos = this.byteBufferRead.position();
            // 开启循环不断读取数据
            while (true) {
                    ......
                    // 如果可读取的字节数大于一个消息头的字节数 + 消息体大小
                    if (diff >= (msgHeaderSize + bodySize)) {
                        byte[] bodyData = new byte[bodySize];
                        this.byteBufferRead.position(this.dispatchPosition + msgHeaderSize);
                        this.byteBufferRead.get(bodyData);
                        // 从读缓冲区中根据消息的位置，读取消息内容，将消息追加到从节点的CommitLog中
                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);

                        this.byteBufferRead.position(readSocketPos);
                        // 更新dispatchPosition的值为消息头大小+消息体大小
                        this.dispatchPosition += msgHeaderSize + bodySize;

                        if (!reportSlaveMaxOffsetPlus()) {
                            return false;
                        }

                        continue;
                    }
                }

                if (!this.byteBufferRead.hasRemaining()) {
                    this.reallocateByteBuffer();
                }

                break;
            }

            return true;
        }

2.4向Master发送主从同步消息拉取偏移量

在HAClient#run()中与主节点建立连接后，会向主节点发送同步消息拉取偏移量，调用reportSlaveMaxOffset()

 private boolean reportSlaveMaxOffset(final long maxOffset) {
            this.reportOffset.position(0);
            this.reportOffset.limit(8); // 设置数据传输大小为8个字节
            this.reportOffset.putLong(maxOffset);// 设置同步偏移量
            this.reportOffset.position(0);
            this.reportOffset.limit(8);

            for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
                try {
                    // 向Master节点发送拉取偏移量
                    this.socketChannel.write(this.reportOffset);
                } catch (IOException e) {
                    log.error(this.getServiceName()
                            + "reportSlaveMaxOffset this.socketChannel.write exception", e);
                    return false;
                }
            }
            // 更新发送时间
            lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
            return !this.reportOffset.hasRemaining();
        }