美文网首页消息中间件
RocketMQ源码-主从同步复制和异步复制

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

作者: persisting_ | 来源:发表于2019-07-23 22:15 被阅读8次


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

    相关文章

      网友评论

        本文标题:RocketMQ源码-主从同步复制和异步复制

        本文链接:https://www.haomeiwen.com/subject/glhvlctx.html