RocketMQ主从同步机制源码解析及关键点总结(二)

梦里梦外; 2022-05-28 07:38 173阅读 0赞

这篇博客接着上篇的讲，主要讲以下两个问题：

1.  Slave接收到Master的数据后的处理
2.  同步/异步Master (SYNC\_MASTER) 下的主从同步机制及其区别

Slave端为SocketChannel注册了SelectionKey.OP\_READ事件，当Master向SocketChannel中写入数据后，Slave能马上感知到，然后将SocketChannel中的数据读出来。

class HAClient extends ServiceThread { 
    
        private boolean connectMaster() throws ClosedChannelException {
            ......
            //SocketChannel注册SelectionKey.OP_READ事件
            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
        }
    
        public void run() {  
            ......
            //最多阻塞1S,直到Master有数据同步于过来。若1S满了还是没有接受到数据,中断阻塞,
            // 执行processReadEvent(),但结果读入byteBufferRead的大小为0,然后循环到这步
            this.selector.select(1000);
    
            // 处理读取事件
            boolean ok = this.processReadEvent();
            if (!ok) {
                this.closeMaster();
            }
    
            // 若进度有变化，上报到Master进度
            if (!reportSlaveMaxOffsetPlus()) {
                continue;
            }
            ......           
        }
    
    }

当Master将数据写入SocketChannel后，Slave马上感知到SocketChannel里有数据，此时中断selector.select(1000)，马上触发processReadEvent( )方法的执行。

/** * 处理从Master读取到数据的事件 * * @return */
    private boolean processReadEvent() {
        int readSizeZeroTimes = 0;
        while (this.byteBufferRead.hasRemaining()) {     // byteBufferRead 还没有读满
            try {
                //将从Master处通过SocketChannel获取到的数据写入 byteBufferRead,返回写入的字节数量
                int readSize = this.socketChannel.read(this.byteBufferRead);
                if (readSize > 0) {
                    //更新最近写入时间
                    lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
    
                    readSizeZeroTimes = 0;
    
                    boolean result = this.dispatchReadRequest();
                    if (!result) {
                        log.error("HAClient, dispatchReadRequest error");
                        return false;
                    }
                } else if (readSize == 0) {
                    if (++readSizeZeroTimes >= 3) {
                        break;
                    }
                } else {
                    // TODO ERROR
                    log.info("HAClient, processReadEvent read socket < 0");
                    return false;
                }
            } catch (IOException e) {
                log.info("HAClient, processReadEvent read socket exception", e);
                return false;
            }
        }
    
        return true;
    }
    
    /** * 读取Master传输的CommitLog数据，并返回是否异常 * 如果读取到数据，写入CommitLog * 异常原因： * 1. Master传输来的数据offset 不等于 Slave的CommitLog数据最大offset * 2. 上报到Master进度失败 * * @return 是否异常 */
    private boolean dispatchReadRequest() {
        final int msgHeaderSize = 8 + 4; // phyoffset + size
        int readSocketPos = this.byteBufferRead.position();
    
        while (true) {
            // 读取到请求
            int diff = this.byteBufferRead.position() - this.dispatchPostion;
            if (diff >= msgHeaderSize) {
                // 读取masterPhyOffset、bodySize。使用dispatchPostion的原因是：处理数据“粘包”导致数据读取不完整。
                long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPostion);
                int bodySize = this.byteBufferRead.getInt(this.dispatchPostion + 8);
                // 校验 Master传输来的数据offset 是否和 Slave的CommitLog数据最大offset 是否相同。
                long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                if (slavePhyOffset != 0) {
                    //数据中断了,Slave的最大Offset匹配不上推送过来的Offset
                    if (slavePhyOffset != masterPhyOffset) {
                        log.error("master pushed offset not equal the max phy offset in slave, SLAVE: "
                            + slavePhyOffset + " MASTER: " + masterPhyOffset);
                        return false;
                    }
                }
                // 读取到消息
                if (diff >= (msgHeaderSize + bodySize)) {
                    // 写入CommitLog
                    byte[] bodyData = new byte[bodySize];
                    this.byteBufferRead.position(this.dispatchPostion + msgHeaderSize);
                    this.byteBufferRead.get(bodyData);
                    HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);
                    // 设置处理到的位置
                    this.byteBufferRead.position(readSocketPos);
                    this.dispatchPostion += msgHeaderSize + bodySize;
                    // 上报到Master进度
                    if (!reportSlaveMaxOffsetPlus()) {
                        return false;
                    }
                    // 继续循环
                    continue;
                }
            }
    
            // 空间写满，重新分配空间
            if (!this.byteBufferRead.hasRemaining()) {
                this.reallocateByteBuffer();
            }
    
            break;
        }
    
        return true;
    }

从dispatchReadRequest( )方法里可以看到，Slave使用dispatchPostion变量来指定每次处理的位置，其目的是为了应对粘包问题。每次提取数据的body部分，追加到CommitLog，当添加成功一次就马上向Master上报此次的进度。

在异步双写模式下，Master为ASYNC\_MASTER，其通过WriteSocketService自动向Slave写入消息，循环写入，当上次写入完成后就紧接着下次。当CommitLog没有新的消息时，WriteSocketService休眠100ms。

在同步双写模式下，Master为SYNC\_MASTER，当一条消息发送到Master时，Master需要等到Slave确定收到了当前消息才会将存储结果返回给生产者。

先看SYNC\_MASTER在存储消息时的额外步骤：

public class CommitLog { 
    
        /** * 添加消息，返回消息结果 * * @param msg 消息 * @return 结果 */
        public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
    
            ......
            // Synchronous write double 如果是 SYNC_MASTER,马上将信息同步至SLAVE;
            if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
                HAService service = this.defaultMessageStore.getHaService();
                if (msg.isWaitStoreMsgOK()) {
                    // 推送到Slave的Offset是否小于这条消息的Offset,且Slave落后Master的进度在允许范围内(256MB)
                    if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
                        //如果是ASYNC_FLUSH
                        if (null == request) {
                            request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
                        }
                        service.putRequest(request);
    
                        // 唤醒WriteSocketService
                        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: " + msg.getTopic() + " tags: "
                                + msg.getTags() + " client address: " + msg.getBornHostString());
                            putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
                        }
                    }
                    // Slave problem
                    else {
                        // Tell the producer, slave not available
                        putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
                    }
                }
            }
        }
    
        public void putRequest(final CommitLog.GroupCommitRequest request) {
            this.groupTransferService.putRequest(request);
        }
    
    }

HAService 向其groupTransferService添加一个GroupCommitRequest，然后唤醒WriteSocketService，因为WriteSocketService在没有消息时会休眠100ms。在唤醒WriteSocketService后，其马上向Slave传输CommitLog的数据(异步)，然后GroupCommitRequest开始等待，等待Slave的进度返回。

我们先看看Master在接收到Slave进度返回时的处理：

class ReadSocketService extends ServiceThread { 
    
        private boolean processReadEvent() {
            ......
            // 设置Slave CommitLog的最大位置
            HAConnection.this.slaveAckOffset = readOffset;
            // 通知目前Slave进度。主要用于Master节点为同步类型的。
            HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);
        }
    }

Master在每接受到Slave的一次进度同步时，均会执行HAService的notifyTransferSome( )方法：

public class HAService { 
    
        /** * 通知slave进度 * * @param offset slave进度 */
        public void notifyTransferSome(final long offset) {
            for (long value = this.push2SlaveMaxOffset.get(); offset > value; ) {
                boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, offset);
                if (ok) {
                    this.groupTransferService.notifyTransferSome();
                    break;
                } else {
                    value = this.push2SlaveMaxOffset.get();
                }
            }
        }
    
    }

在HAService 的notifyTransferSome( ) 方法里更新push2SlaveMaxOffset的值，然后执行groupTransferService\#notifyTransferSome( )：

/** * GroupTransferService Service */
    class GroupTransferService extends ServiceThread {
    
        public void run() {
            log.info(this.getServiceName() + " service started");
    
            while (!this.isStopped()) {
                try {
                    this.waitForRunning(10);
                    this.doWaitTransfer();
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            log.info(this.getServiceName() + " service end");
        }
    
        /** * 通知slave进度。唤醒等待锁。 */
        public void notifyTransferSome() {
            this.notifyTransferObject.wakeup();
        }
    
        private void doWaitTransfer() {
            synchronized (this.requestsRead) {
                if (!this.requestsRead.isEmpty()) {
                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {
                        // 推送到Slave的Offset是否 >= 当前消息的Offset
                        boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                        //每次从Slave的进度提交请求能够中断wait
                        //最多等5次Slave的Ack,如果Ack的进度 >= 当前消息的进度,则返回true
                        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());
                        }
    
                        // 唤醒GroupCommitRequest，并设置Slave是否同步成功
                        req.wakeupCustomer(transferOK);
                    }
    
                    this.requestsRead.clear();
                }
            }
        }
    
    }
    
    
    public static class GroupCommitRequest { 
    
        /** 请求等待,然后返回Slave是否同步成功 * @param timeout * @return */
        public boolean waitForFlush(long timeout) {
            try {
                this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
                return this.flushOK;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
        }
    
        /** * 终止等待 * * @param flushOK */
        public void wakeupCustomer(final boolean flushOK) {
            this.flushOK = flushOK;
            this.countDownLatch.countDown();
        }
    
    
    }

HAService在启动时也会启动GroupTransferService ，GroupTransferService 在启动后会循环执行doWaitTransfer( )方法，循环检测GroupCommitRequest 的执行状态。每存储一条消息生成一个GroupCommitRequest ，加入GroupTransferService 。GroupTransferService 每10ms对其持有的GroupCommitRequest 集合做响应处理。Slave的每次上报进度唤醒GroupTransferService 的等待，当发现上报的进度大于等于当前消息的存储进度时，唤醒GroupCommitRequest 的waitForFlush( )；当Slave的5次进度上报都小于当前消息的存储进度时，也唤醒GroupCommitRequest 的waitForFlush( )。唤醒的同时设置Slave是否Flush成功。

唤醒之后，CommitLog就能继续消息的处理流程了，将Slave的Flush结果返回给消费者。

总结：

1.  同步双写模式下，消息存储时需要等到Slave的进度上报给Master，最多等待5次上报，然后根据上报的进度是否大于等于当前消息的进度来决定同步的结果，若同步失败，将结果返回给消费者。
2.  Slave每接收到一次数据就上报一次自身进度