RocketMQ：主從同步的實(shí)現(xiàn)

#所屬集群名字brokerClusterName=DefaultCluster#broker名字，名字可重復(fù),為了管理,每個(gè)master起一個(gè)名字,他的slave同他,eg:Amaster叫broker-a,他的slave也叫broker-abrokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=0#Broker 的角色#- ASYNC_MASTER 異步復(fù)制Master#- SYNC_MASTER 同步雙寫(xiě)Master#- SLAVEbrokerRole=ASYNC_MASTER#刷盤方式#- ASYNC_FLUSH 異步刷盤#- SYNC_FLUSH 同步刷盤flushDiskType=ASYNC_FLUSH#nameServer地址，分號(hào)分割namesrvAddr=172.0.1.5:9876;172.0.1.6:9876#在發(fā)送消息時(shí)，自動(dòng)創(chuàng)建服務(wù)器不存在的topic，默認(rèn)創(chuàng)建的隊(duì)列數(shù)defaultTopicQueueNums=4#是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開(kāi)啟，線上關(guān)閉autoCreateTopicEnable=true#是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開(kāi)啟，線上關(guān)閉autoCreateSubscriptionGroup=true#Broker 對(duì)外服務(wù)的監(jiān)聽(tīng)端口,listenPort=10911#刪除文件時(shí)間點(diǎn)，默認(rèn)凌晨 4點(diǎn)deleteWhen=04#文件保留時(shí)間，默認(rèn) 48 小時(shí)fileReservedTime=120#commitLog每個(gè)文件的大小默認(rèn)1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每個(gè)文件默認(rèn)存30W條，根據(jù)業(yè)務(wù)情況調(diào)整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#檢測(cè)物理文件磁盤空間diskMaxUsedSpaceRatio=88#存儲(chǔ)路徑storePathRootDir=/usr/local/rocketmq/store/broker-a#commitLog 存儲(chǔ)路徑storePathCommitLog=/usr/local/rocketmq/store/broker-a/commitlog#消費(fèi)隊(duì)列存儲(chǔ)路徑存儲(chǔ)路徑storePathConsumeQueue=/usr/local/rocketmq/store/broker-a/consumequeue#消息索引存儲(chǔ)路徑storePathIndex=/usr/local/rocketmq/store/broker-a/index#checkpoint 文件存儲(chǔ)路徑storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存儲(chǔ)路徑abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#checkTransactionMessageEnable=false#發(fā)消息線程池?cái)?shù)量#sendMessageThreadPoolNums=128#拉消息線程池?cái)?shù)量#pullMessageThreadPoolNums=128

#所屬集群名字brokerClusterName=DefaultCluster#broker名字，名字可重復(fù),為了管理,每個(gè)master起一個(gè)名字,他的slave同他,eg:Amaster叫broker-a,他的slave也叫broker-abrokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=1#Broker 的角色#- ASYNC_MASTER 異步復(fù)制Master#- SYNC_MASTER 同步雙寫(xiě)Master#- SLAVEbrokerRole=SLAVE#刷盤方式#- ASYNC_FLUSH 異步刷盤#- SYNC_FLUSH 同步刷盤flushDiskType=ASYNC_FLUSH#nameServer地址，分號(hào)分割namesrvAddr=172.0.1.5:9876;172.0.1.6:9876#在發(fā)送消息時(shí)，自動(dòng)創(chuàng)建服務(wù)器不存在的topic，默認(rèn)創(chuàng)建的隊(duì)列數(shù)defaultTopicQueueNums=4#是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開(kāi)啟，線上關(guān)閉autoCreateTopicEnable=true#是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開(kāi)啟，線上關(guān)閉autoCreateSubscriptionGroup=true#Broker 對(duì)外服務(wù)的監(jiān)聽(tīng)端口,listenPort=10920#刪除文件時(shí)間點(diǎn)，默認(rèn)凌晨 4點(diǎn)deleteWhen=04#文件保留時(shí)間，默認(rèn) 48 小時(shí)fileReservedTime=120#commitLog每個(gè)文件的大小默認(rèn)1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每個(gè)文件默認(rèn)存30W條，根據(jù)業(yè)務(wù)情況調(diào)整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#檢測(cè)物理文件磁盤空間diskMaxUsedSpaceRatio=88#存儲(chǔ)路徑storePathRootDir=/usr/local/rocketmq/store/broker-a-s#commitLog 存儲(chǔ)路徑storePathCommitLog=/usr/local/rocketmq/store/broker-a-s/commitlog#消費(fèi)隊(duì)列存儲(chǔ)路徑存儲(chǔ)路徑storePathConsumeQueue=/usr/local/rocketmq/store/broker-a-s/consumequeue#消息索引存儲(chǔ)路徑storePathIndex=/usr/local/rocketmq/store/broker-a-s/index#checkpoint 文件存儲(chǔ)路徑storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存儲(chǔ)路徑abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#checkTransactionMessageEnable=false#發(fā)消息線程池?cái)?shù)量#sendMessageThreadPoolNums=128#拉消息線程池?cái)?shù)量#pullMessageThreadPoolNums=128

  nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties > logs/broker-a.log 2>&1 &    nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties > logs/broker-a-s.log 2>&1 &

// org.apache.rocketmq.broker.BrokerController#start    public void start() throws Exception {        if (this.messageStore != null) {            this.messageStore.start();        }
        if (this.remotingServer != null) {            this.remotingServer.start();        }
        if (this.fastRemotingServer != null) {            this.fastRemotingServer.start();        }
        if (this.fileWatchService != null) {            this.fileWatchService.start();        }
        if (this.brokerOuterAPI != null) {            this.brokerOuterAPI.start();        }
        if (this.pullRequestHoldService != null) {            this.pullRequestHoldService.start();        }
        if (this.clientHousekeepingService != null) {            this.clientHousekeepingService.start();        }
        if (this.filterServerManager != null) {            this.filterServerManager.start();        }
        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            startProcessorByHa(messageStoreConfig.getBrokerRole());            // 處理SLAVE消息同步            handleSlaveSynchronize(messageStoreConfig.getBrokerRole());            // 強(qiáng)制做一次注冊(cè)動(dòng)作            this.registerBrokerAll(true, false, true);        }        // 定期向nameserver注冊(cè)自身狀態(tài)        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override            public void run() {                try {                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());                } catch (Throwable e) {                    log.error("registerBrokerAll Exception", e);                }            }        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
        if (this.brokerStatsManager != null) {            this.brokerStatsManager.start();        }
        if (this.brokerFastFailure != null) {            this.brokerFastFailure.start();        }

    }
    private void handleSlaveSynchronize(BrokerRole role) {        // 只有slave節(jié)點(diǎn)，才進(jìn)行同步操作        if (role == BrokerRole.SLAVE) {            if (null != slaveSyncFuture) {                slaveSyncFuture.cancel(false);            }            // 設(shè)置master節(jié)點(diǎn)為空，避免一開(kāi)始就進(jìn)行同步            // 后續(xù)必然有其他地方設(shè)計(jì) master信息            // 實(shí)際上它是在registerBrokerAll() 的時(shí)候，將master信息放入的            this.slaveSynchronize.setMasterAddr(null);            // 10秒鐘同步一次數(shù)據(jù)            slaveSyncFuture = this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {                @Override                public void run() {                    try {                        BrokerController.this.slaveSynchronize.syncAll();                    }                    catch (Throwable e) {                        log.error("ScheduledTask SlaveSynchronize syncAll error.", e);                    }                }            }, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS);        } else {            //handle the slave synchronise            if (null != slaveSyncFuture) {                slaveSyncFuture.cancel(false);            }            this.slaveSynchronize.setMasterAddr(null);        }    }    public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {        TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
        if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())            || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {            ConcurrentHashMap topicConfigTable = new ConcurrentHashMap();            for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {                TopicConfig tmp =                    new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),                        this.brokerConfig.getBrokerPermission());                topicConfigTable.put(topicConfig.getTopicName(), tmp);            }            topicConfigWrapper.setTopicConfigTable(topicConfigTable);        }        // 強(qiáng)制注冊(cè)或者進(jìn)行周期性注冊(cè)時(shí)間到時(shí)，向nameserver注冊(cè)自身        if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),                                        this.getBrokerAddr(),                                        this.brokerConfig.getBrokerName(),                                        this.brokerConfig.getBrokerId(),                                        this.brokerConfig.getRegisterBrokerTimeoutMills())) {            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);        }    }
    private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,        TopicConfigSerializeWrapper topicConfigWrapper) {        // 向多個(gè)nameserver依次注冊(cè)        List registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(            this.brokerConfig.getBrokerClusterName(),            this.getBrokerAddr(),            this.brokerConfig.getBrokerName(),            this.brokerConfig.getBrokerId(),            this.getHAServerAddr(),            topicConfigWrapper,            this.filterServerManager.buildNewFilterServerList(),            oneway,            this.brokerConfig.getRegisterBrokerTimeoutMills(),            this.brokerConfig.isCompressedRegister());
        if (registerBrokerResultList.size() > 0) {            RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);            if (registerBrokerResult != null) {                if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {                    this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());                }                // 更新master地址信息                this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
                if (checkOrderConfig) {                    this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());                }            }        }    }    // org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBrokerAll    public List registerBrokerAll(        final String clusterName,        final String brokerAddr,        final String brokerName,        final long brokerId,        final String haServerAddr,        final TopicConfigSerializeWrapper topicConfigWrapper,        final List filterServerList,        final boolean oneway,        final int timeoutMills,        final boolean compressed) {
        final List registerBrokerResultList = Lists.newArrayList();        List nameServerAddressList = this.remotingClient.getNameServerAddressList();        if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
            final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();            requestHeader.setBrokerAddr(brokerAddr);            requestHeader.setBrokerId(brokerId);            requestHeader.setBrokerName(brokerName);            requestHeader.setClusterName(clusterName);            requestHeader.setHaServerAddr(haServerAddr);            requestHeader.setCompressed(compressed);
            RegisterBrokerBody requestBody = new RegisterBrokerBody();            requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);            requestBody.setFilterServerList(filterServerList);            final byte[] body = requestBody.encode(compressed);            final int bodyCrc32 = UtilAll.crc32(body);            requestHeader.setBodyCrc32(bodyCrc32);            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());            for (final String namesrvAddr : nameServerAddressList) {                // 多線程同時(shí)注冊(cè)多個(gè)nameserver, 效果更佳                brokerOuterExecutor.execute(new Runnable() {                    @Override                    public void run() {                        try {                            RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);                            if (result != null) {                                registerBrokerResultList.add(result);                            }
                            log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);                        } catch (Exception e) {                            log.warn("registerBroker Exception, {}", namesrvAddr, e);                        } finally {                            countDownLatch.countDown();                        }                    }                });            }
            try {                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);            } catch (InterruptedException e) {            }        }
        return registerBrokerResultList;    }

// org.apache.rocketmq.broker.slave.SlaveSynchronize#syncAll    public void syncAll() {        // 同步topic配置信息        this.syncTopicConfig();        // 同步消費(fèi)偏移量信息        this.syncConsumerOffset();        // 同步延遲信息        this.syncDelayOffset();        // 同步消費(fèi)組信息數(shù)據(jù)，所以主從同步的核心，是基于消息的訂閱來(lái)實(shí)現(xiàn)的        this.syncSubscriptionGroupConfig();    }
    // 同步topic配置信息    private void syncTopicConfig() {        String masterAddrBak = this.masterAddr;        // 存在master地址，且該地址不是自身時(shí)，才會(huì)進(jìn)行同步動(dòng)作        if (masterAddrBak != null && !masterAddrBak.equals(brokerController.getBrokerAddr())) {            try {                TopicConfigSerializeWrapper topicWrapper =                    this.brokerController.getBrokerOuterAPI().getAllTopicConfig(masterAddrBak);                // 版本發(fā)生變更，即數(shù)據(jù)有變化，則寫(xiě)入新的版本數(shù)據(jù)                if (!this.brokerController.getTopicConfigManager().getDataVersion()                    .equals(topicWrapper.getDataVersion())) {
                    this.brokerController.getTopicConfigManager().getDataVersion()                        .assignNewOne(topicWrapper.getDataVersion());                    this.brokerController.getTopicConfigManager().getTopicConfigTable().clear();                    this.brokerController.getTopicConfigManager().getTopicConfigTable()                        .putAll(topicWrapper.getTopicConfigTable());                    // 持久化topic信息                    this.brokerController.getTopicConfigManager().persist();
                    log.info("Update slave topic config from master, {}", masterAddrBak);                }            } catch (Exception e) {                log.error("SyncTopicConfig Exception, {}", masterAddrBak, e);            }        }    }
    // 同步消費(fèi)偏移量信息    private void syncConsumerOffset() {        String masterAddrBak = this.masterAddr;        if (masterAddrBak != null && !masterAddrBak.equals(brokerController.getBrokerAddr())) {            try {                ConsumerOffsetSerializeWrapper offsetWrapper =                    this.brokerController.getBrokerOuterAPI().getAllConsumerOffset(masterAddrBak);                this.brokerController.getConsumerOffsetManager().getOffsetTable()                    .putAll(offsetWrapper.getOffsetTable());                this.brokerController.getConsumerOffsetManager().persist();                log.info("Update slave consumer offset from master, {}", masterAddrBak);            } catch (Exception e) {                log.error("SyncConsumerOffset Exception, {}", masterAddrBak, e);            }        }    }    // 額。。。 反正就是一個(gè)數(shù)字吧, 存儲(chǔ)在 config/delayOffset.json 下    private void syncDelayOffset() {        String masterAddrBak = this.masterAddr;        if (masterAddrBak != null && !masterAddrBak.equals(brokerController.getBrokerAddr())) {            try {                String delayOffset =                    this.brokerController.getBrokerOuterAPI().getAllDelayOffset(masterAddrBak);                if (delayOffset != null) {
                    String fileName =                        StorePathConfigHelper.getDelayOffsetStorePath(this.brokerController                            .getMessageStoreConfig().getStorePathRootDir());                    try {                        MixAll.string2File(delayOffset, fileName);                    } catch (IOException e) {                        log.error("Persist file Exception, {}", fileName, e);                    }                }                log.info("Update slave delay offset from master, {}", masterAddrBak);            } catch (Exception e) {                log.error("SyncDelayOffset Exception, {}", masterAddrBak, e);            }        }    }
    // 同步消費(fèi)組信息數(shù)據(jù)    private void syncSubscriptionGroupConfig() {        String masterAddrBak = this.masterAddr;        if (masterAddrBak != null  && !masterAddrBak.equals(brokerController.getBrokerAddr())) {            try {                SubscriptionGroupWrapper subscriptionWrapper =                    this.brokerController.getBrokerOuterAPI()                        .getAllSubscriptionGroupConfig(masterAddrBak);
                if (!this.brokerController.getSubscriptionGroupManager().getDataVersion()                    .equals(subscriptionWrapper.getDataVersion())) {                    SubscriptionGroupManager subscriptionGroupManager =                        this.brokerController.getSubscriptionGroupManager();                    subscriptionGroupManager.getDataVersion().assignNewOne(                        subscriptionWrapper.getDataVersion());                    subscriptionGroupManager.getSubscriptionGroupTable().clear();                    subscriptionGroupManager.getSubscriptionGroupTable().putAll(                        subscriptionWrapper.getSubscriptionGroupTable());                    // 持久化消費(fèi)組信息                    subscriptionGroupManager.persist();                    log.info("Update slave Subscription Group from master, {}", masterAddrBak);                }            } catch (Exception e) {                log.error("SyncSubscriptionGroup Exception, {}", masterAddrBak, e);            }        }    }

// org.apache.rocketmq.store.DefaultMessageStore#DefaultMessageStore    public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {        this.messageArrivingListener = messageArrivingListener;        this.brokerConfig = brokerConfig;        this.messageStoreConfig = messageStoreConfig;        this.brokerStatsManager = brokerStatsManager;        this.allocateMappedFileService = new AllocateMappedFileService(this);        if (messageStoreConfig.isEnableDLegerCommitLog()) {            this.commitLog = new DLedgerCommitLog(this);        } else {            this.commitLog = new CommitLog(this);        }        this.consumeQueueTable = new ConcurrentHashMap<>(32);
        this.flushConsumeQueueService = new FlushConsumeQueueService();        this.cleanCommitLogService = new CleanCommitLogService();        this.cleanConsumeQueueService = new CleanConsumeQueueService();        this.storeStatsService = new StoreStatsService();        this.indexService = new IndexService(this);        if (!messageStoreConfig.isEnableDLegerCommitLog()) {            // 初始化 HAService            this.haService = new HAService(this);        } else {            this.haService = null;        }        ...        File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));        MappedFile.ensureDirOK(file.getParent());        lockFile = new RandomAccessFile(file, "rw");    }
    // org.apache.rocketmq.store.ha.HAService#HAService    public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {        this.defaultMessageStore = defaultMessageStore;        // 開(kāi)啟server端服務(wù)        this.acceptSocketService =            new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());        this.groupTransferService = new GroupTransferService();        // 初始化client        this.haClient = new HAClient();    }    // 具體運(yùn)行則都會(huì)被視為一個(gè)個(gè)的后臺(tái)線程，會(huì)在start()操作中統(tǒng)一運(yùn)行起來(lái)    public void start() throws Exception {        // server端服務(wù)啟動(dòng)，由master節(jié)點(diǎn)管控        this.acceptSocketService.beginAccept();        this.acceptSocketService.start();        // 數(shù)據(jù)中轉(zhuǎn)服務(wù)，它會(huì)接收用戶的寫(xiě)請(qǐng)求，然后吐數(shù)據(jù)給到各slave節(jié)點(diǎn)        this.groupTransferService.start();        // 客戶端請(qǐng)求服務(wù)，由slave節(jié)點(diǎn)發(fā)起        this.haClient.start();    }

// org.apache.rocketmq.store.ha.HAService.HAClient#run        @Override        public void run() {            log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {                try {                    // 使用原生nio, 嘗試連接至master                    if (this.connectMaster()) {
                        if (this.isTimeToReportOffset()) {                            // 隔一段時(shí)間向master匯報(bào)一次本slave的同步信息                            boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);                            // 如果連接無(wú)效，則關(guān)閉，下次再循環(huán)周期將會(huì)重新發(fā)起連接                            if (!result) {                                this.closeMaster();                            }                        }                        this.selector.select(1000);                        // 核心邏輯：處理獲取到的消息數(shù)據(jù)                        boolean ok = this.processReadEvent();                        if (!ok) {                            this.closeMaster();                        }
                        if (!reportSlaveMaxOffsetPlus()) {                            continue;                        }
                        long interval =                            HAService.this.getDefaultMessageStore().getSystemClock().now()                                - this.lastWriteTimestamp;                        if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig()                            .getHaHousekeepingInterval()) {                            log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress                                + "] expired, " + interval);                            this.closeMaster();                            log.warn("HAClient, master not response some time, so close connection");                        }                    } else {                        // 未連接成功，5秒后重試，可能會(huì)一直無(wú)用                        this.waitForRunning(1000 * 5);                    }                } catch (Exception e) {                    log.warn(this.getServiceName() + " service has exception. ", e);                    this.waitForRunning(1000 * 5);                }            }
            log.info(this.getServiceName() + " service end");        }
        private boolean connectMaster() throws ClosedChannelException {            // 單例長(zhǎng)鏈接            if (null == socketChannel) {                String addr = this.masterAddress.get();                // 如果沒(méi)有master, 則返回空                // 針對(duì)master節(jié)點(diǎn)，也是同樣的運(yùn)行，只是不會(huì)連接到任何節(jié)點(diǎn)而已                if (addr != null) {
                    SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);                    if (socketAddress != null) {                        // 原生nio實(shí)現(xiàn)                        this.socketChannel = RemotingUtil.connect(socketAddress);                        if (this.socketChannel != null) {                            this.socketChannel.register(this.selector, SelectionKey.OP_READ);                        }                    }                }
                this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                this.lastWriteTimestamp = System.currentTimeMillis();            }
            return this.socketChannel != null;        }    // org.apache.rocketmq.remoting.common.RemotingUtil#connect    public static SocketChannel connect(SocketAddress remote) {        return connect(remote, 1000 * 5);    }    public static SocketChannel connect(SocketAddress remote, final int timeoutMillis) {        SocketChannel sc = null;        try {            sc = SocketChannel.open();            sc.configureBlocking(true);            sc.socket().setSoLinger(false, -1);            sc.socket().setTcpNoDelay(true);            sc.socket().setReceiveBufferSize(1024 * 64);            sc.socket().setSendBufferSize(1024 * 64);            sc.socket().connect(remote, timeoutMillis);            sc.configureBlocking(false);            return sc;        } catch (Exception e) {            if (sc != null) {                try {                    sc.close();                } catch (IOException e1) {                    e1.printStackTrace();                }            }        }        return null;    }    processReadEvent() 即是在收到master的新數(shù)據(jù)后，實(shí)現(xiàn)如何同步到本broker的commitlog中。其實(shí)現(xiàn)主要還是依賴于commitlogService.        // org.apache.rocketmq.store.ha.HAService.HAClient#processReadEvent        private boolean processReadEvent() {            int readSizeZeroTimes = 0;            while (this.byteBufferRead.hasRemaining()) {                try {                    int readSize = this.socketChannel.read(this.byteBufferRead);                    if (readSize > 0) {                        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;        }
        private boolean dispatchReadRequest() {            // 按協(xié)議讀取數(shù)據(jù)            final int msgHeaderSize = 8 + 4; // phyoffset + size            int readSocketPos = this.byteBufferRead.position();
            while (true) {                int diff = this.byteBufferRead.position() - this.dispatchPosition;                if (diff >= msgHeaderSize) {                    long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPosition);                    int bodySize = this.byteBufferRead.getInt(this.dispatchPosition + 8);
                    long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                    if (slavePhyOffset != 0) {                        if (slavePhyOffset != masterPhyOffset) {                            log.error("master pushed offset not equal the max phy offset in slave, SLAVE: "                                + slavePhyOffset + " MASTER: " + masterPhyOffset);                            return false;                        }                    }                    // 數(shù)據(jù)讀取完成，則立即添加到存儲(chǔ)中                    if (diff >= (msgHeaderSize + bodySize)) {                        byte[] bodyData = new byte[bodySize];                        this.byteBufferRead.position(this.dispatchPosition + msgHeaderSize);                        this.byteBufferRead.get(bodyData);
                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);
                        this.byteBufferRead.position(readSocketPos);                        this.dispatchPosition += msgHeaderSize + bodySize;
                        if (!reportSlaveMaxOffsetPlus()) {                            return false;                        }
                        continue;                    }                }
                if (!this.byteBufferRead.hasRemaining()) {                    this.reallocateByteBuffer();                }
                break;            }
            return true;        }    // org.apache.rocketmq.store.DefaultMessageStore#appendToCommitLog    @Override    public boolean appendToCommitLog(long startOffset, byte[] data) {        if (this.shutdown) {            log.warn("message store has shutdown, so appendToPhyQueue is forbidden");            return false;        }        // 添加到commitlog中，并生成后續(xù)的consumeQueue,index等相關(guān)信息        boolean result = this.commitLog.appendData(startOffset, data);        if (result) {            this.reputMessageService.wakeup();        } else {            log.error("appendToPhyQueue failed " + startOffset + " " + data.length);        }
        return result;    }

// org.apache.rocketmq.store.ha.HAService.AcceptSocketService    /**     * Listens to slave connections to create {@link HAConnection}.     */    class AcceptSocketService extends ServiceThread {        private final SocketAddress socketAddressListen;        private ServerSocketChannel serverSocketChannel;        private Selector selector;        // 給定端口監(jiān)聽(tīng)        public AcceptSocketService(final int port) {            this.socketAddressListen = new InetSocketAddress(port);        }
        /**         * Starts listening to slave connections.         *         * @throws Exception If fails.         */        public void beginAccept() throws Exception {            this.serverSocketChannel = ServerSocketChannel.open();            this.selector = RemotingUtil.openSelector();            this.serverSocketChannel.socket().setReuseAddress(true);            this.serverSocketChannel.socket().bind(this.socketAddressListen);            this.serverSocketChannel.configureBlocking(false);            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);        }
        /**         * {@inheritDoc}         */        @Override        public void run() {            log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {                try {                    this.selector.select(1000);                    Set 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 conn = new HAConnection(HAService.this, sc);                                        // accept 接入后，開(kāi)啟另外的讀線程處理數(shù)據(jù)請(qǐng)求                                        conn.start();                                        HAService.this.addConnection(conn);                                    } catch (Exception e) {                                        log.error("new HAConnection exception", e);                                        sc.close();                                    }                                }                            } else {                                log.warn("Unexpected ops in select " + k.readyOps());                            }                        }
                        selected.clear();                    }                } catch (Exception e) {                    log.error(this.getServiceName() + " service has exception.", e);                }            }
            log.info(this.getServiceName() + " service end");        }        ...    }    // org.apache.rocketmq.store.ha.HAConnection#start    public void start() {        this.readSocketService.start();        this.writeSocketService.start();    }        // org.apache.rocketmq.store.ha.HAConnection.ReadSocketService#run        @Override        public void run() {            HAConnection.log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {                try {                    this.selector.select(1000);                    boolean ok = this.processReadEvent();                    if (!ok) {                        HAConnection.log.error("processReadEvent error");                        break;                    }
                    long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;                    if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {                        log.warn("ha housekeeping, found this connection[" + HAConnection.this.clientAddr + "] expired, " + interval);                        break;                    }                } catch (Exception e) {                    HAConnection.log.error(this.getServiceName() + " service has exception.", e);                    break;                }            }
            this.makeStop();
            writeSocketService.makeStop();
            haService.removeConnection(HAConnection.this);
            HAConnection.this.haService.getConnectionCount().decrementAndGet();
            SelectionKey sk = this.socketChannel.keyFor(this.selector);            if (sk != null) {                sk.cancel();            }
            try {                this.selector.close();                this.socketChannel.close();            } catch (IOException e) {                HAConnection.log.error("", e);            }
            HAConnection.log.info(this.getServiceName() + " service end");        }
        private boolean processReadEvent() {            int readSizeZeroTimes = 0;
            if (!this.byteBufferRead.hasRemaining()) {                this.byteBufferRead.flip();                this.processPosition = 0;            }
            while (this.byteBufferRead.hasRemaining()) {                try {                    int readSize = this.socketChannel.read(this.byteBufferRead);                    if (readSize > 0) {                        readSizeZeroTimes = 0;                        this.lastReadTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();                        if ((this.byteBufferRead.position() - this.processPosition) >= 8) {                            int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);                            long readOffset = this.byteBufferRead.getLong(pos - 8);                            this.processPosition = pos;                            // 讀取唯一參數(shù)                            HAConnection.this.slaveAckOffset = readOffset;                            if (HAConnection.this.slaveRequestOffset < 0) {                                HAConnection.this.slaveRequestOffset = readOffset;                                log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);                            }                            // ...                            HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);                        }                    } else if (readSize == 0) {                        if (++readSizeZeroTimes >= 3) {                            break;                        }                    } else {                        log.error("read socket[" + HAConnection.this.clientAddr + "] < 0");                        return false;                    }                } catch (IOException e) {                    log.error("processReadEvent exception", e);                    return false;                }            }
            return true;        }    // org.apache.rocketmq.store.ha.HAService#notifyTransferSome    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();            }        }    }

// org.apache.rocketmq.store.ha.HAService.GroupTransferService#putRequest        public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {            synchronized (this.requestsWrite) {                this.requestsWrite.add(request);            }            this.wakeup();        }
        public void notifyTransferSome() {            this.notifyTransferObject.wakeup();        }
        private void swapRequests() {            // 交換buffer            List tmp = this.requestsWrite;            this.requestsWrite = this.requestsRead;            this.requestsRead = tmp;        }
        private void doWaitTransfer() {            synchronized (this.requestsRead) {                if (!this.requestsRead.isEmpty()) {                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {                        boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();                        long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()                            + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();                        while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {                            this.notifyTransferObject.waitForRunning(1000);                            transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();                        }
                        if (!transferOK) {                            log.warn("transfer messsage to slave timeout, " + req.getNextOffset());                        }
                        req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT);                    }
                    this.requestsRead.clear();                }            }        }
        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");        }