7、Broker消息刷盘以及主从同步 (七)

素颜马尾好姑娘i 2024-02-05 12:13 8阅读 0赞

今天我们主要来分析消息的刷盘以及主从同步，简单理解为Producer将消息发给Broker，Broker是如何将消息保存到磁盘里面，以及当我们搭建主从架构时又是如何通知Slave节点。

![在这里插入图片描述][c79e3c6ac47c4d80939c6e9030d715bd.png]  
消息刷盘

前面消息处理执行CommitLog的asyncPutMessage()将消息存储实际上显示存放在内存里面，最后还有一个submitFlushRequest方法进行刷盘申请。

public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
        
        // 同步刷盘
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        
            final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
            if (messageExt.isWaitStoreMsgOK()) {
        
                GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                        this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
                flushDiskWatcher.add(request);
                // 添加完后就执行run方法
                service.putRequest(request);
                return request.future();
            } else {
        
                service.wakeup();
                return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
            }
        }
        // 异步存盘
        else {
        
            // isTransientStorePoolEnable是否启用堆外内存
            if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
        
                // 主动flush操作（执行run方法）
                flushCommitLogService.wakeup();
            } else  {
        
                // 不会主动操作，操作系统
                commitLogService.wakeup();
            }
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }

两种刷盘机制：同步刷盘和异步刷盘

同步刷盘  
只有真正持久化到磁盘里面Broker端才会向Producer端发送一个成功的ACK响应，该模式比较严谨与可靠，但是性能方法大打折扣。

class GroupCommitService extends FlushCommitLogService {
        
        // 写刷盘请求
        private volatile LinkedList<GroupCommitRequest> requestsWrite = new LinkedList<GroupCommitRequest>();
        // 读刷盘请求
        private volatile LinkedList<GroupCommitRequest> requestsRead = new LinkedList<GroupCommitRequest>();
        private final PutMessageSpinLock lock = new PutMessageSpinLock();
    
        public synchronized void putRequest(final GroupCommitRequest request) {
        
            lock.lock();
            try {
        
                // 存入写刷盘请求
                this.requestsWrite.add(request);
            } finally {
        
                lock.unlock();
            }
            this.wakeup();
        }
      // 读写备份
        private void swapRequests() {
        
            lock.lock();
            try {
        
                // 进行备份
                LinkedList<GroupCommitRequest> tmp = this.requestsWrite;
                this.requestsWrite = this.requestsRead;
                this.requestsRead = tmp;
            } finally {
        
                lock.unlock();
            }
        }
        // 使用读刷盘请求刷新磁盘
        private void doCommit() {
        
            // 读队列和写队列（requestsWrite写入，requestsRead读取）
            if (!this.requestsRead.isEmpty()) {
        
                for (GroupCommitRequest req : this.requestsRead) {
        
                    // There may be a message in the next file, so a maximum of
                    // two times the flush
                    // 判断是否刷新成功
                    boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                    for (int i = 0; i < 2 && !flushOK; i++) {
        
                        // 刷盘 执行强制刷盘操作，最少刷0页，即所有消息都会刷盘
                        CommitLog.this.mappedFileQueue.flush(0);
                        // 判断是否刷新成功，一个文件的内存不足就需要在执行一遍刷盘
                        flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                    }
    
                    req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
                }
    
                long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
                if (storeTimestamp > 0) {
        
                    CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
                }
    
                this.requestsRead = new LinkedList<>();
            } else {
        
                // Because of individual messages is set to not sync flush, it
                // will come to this process
                CommitLog.this.mappedFileQueue.flush(0);
            }
        }
      // 执行刷盘run方法
        public void run() {
        
            CommitLog.log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {
        
                try {
        
                    // 休眠10s进行刷盘（现在百分之百决定安全）
                    this.waitForRunning(10);
                    // 提交刷盘
                    this.doCommit();
                } catch (Exception e) {
        
                    CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
    
            // Under normal circumstances shutdown, wait for the arrival of the
            // request, and then flush
            try {
        
                Thread.sleep(10);
            } catch (InterruptedException e) {
        
                CommitLog.log.warn(this.getServiceName() + " Exception, ", e);
            }
    
            synchronized (this) {
        
                // 读写备份
                this.swapRequests();
            }
    
            this.doCommit();
    
            CommitLog.log.info(this.getServiceName() + " service end");
        }
        // ...
    }

同步刷盘的类实现GroupCommitService，在创建该类的时候会初始化两个队列，一个负责读取刷盘请求，一个负责写入刷盘请求，同时初始化一个独占锁保证数据的安全性，读写分离消费刷盘请求之间锁竞争。  
GroupCommitService类是一个线程类，当我们执行它的start方法时就会来执行run方法，run方法的逻辑就是去刷盘，只要服务没停一直执行（休眠10s，备份Write队列到Read队列，执行刷盘操作）。  
执行doCommit真正执行刷盘操作，当读刷盘请求存在数据的时候就会去进行数据刷盘，由于会存在一个文件可能存放不下所以在刷盘的时候最多会刷新两个，如果没存在数据也会去刷盘，刷盘的数据为0不会刷数据而会更新刷盘时间戳。  
注意：尽管是同步刷盘，但是还是会存在10s的延时时间。

异步刷盘  
只要消息写入缓存就会往Producer发送ACK响应，该方式降低读写延迟，提高吞吐量。

堆内存模式：根据FlushRealTimeService类的run方法进行异步刷盘。

该方法里面存在很多可以配合的参数，比如：是否支持定时刷盘默认false（flushCommitLogTimed进行配置），刷盘间隔时间默认500ms（flushIntervalCommitLog进行配置），刷盘的最小页数默认4即16k（flushCommitLogLeastPages进行配置），最长延迟刷盘时间默认10s（flushCommitLogThoroughInterval进行配置）。

如果当刷盘时间距离上一次大于10s，那么就不管是否大于4直接就会去刷盘，更新刷盘时间为当前时间。

最后当正常关闭后还是会去执行10次刷盘操作，这样保证数据的准确性，尽量让消息少丢失。

堆外内存模式：根据CommitRealTimeService类的run方法来进行异步刷盘。

public void run() {
        
        CommitLog.log.info(this.getServiceName() + " service started");
        while (!this.isStopped()) {
        
            // 提交时间间隔默认200ms
            int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
          // 刷盘最小页数 默认4
            int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
            // 最长刷盘间隔时间，默认200ms
            int commitDataThoroughInterval =
                CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();
    
            long begin = System.currentTimeMillis();
            if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
        
                this.lastCommitTimestamp = begin;
                commitDataLeastPages = 0;
            }
    
            try {
        
                // 直接提交给操作系统的PageCache,写入文件由操作系统去操作
                boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
                long end = System.currentTimeMillis();
                if (!result) {
        
                    this.lastCommitTimestamp = end; // result = false means some data committed.
                    //now wake up flush thread.
                    flushCommitLogService.wakeup();
                }
    
                if (end - begin > 500) {
        
                    log.info("Commit data to file costs {} ms", end - begin);
                }
                this.waitForRunning(interval);
            } catch (Throwable e) {
        
                CommitLog.log.error(this.getServiceName() + " service has exception. ", e);
            }
        }
    
        boolean result = false;
    
      // 正常停止后在执行10次刷盘
        for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
        
            result = CommitLog.this.mappedFileQueue.commit(0);
            CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
        }
        CommitLog.log.info(this.getServiceName() + " service end");
    }

该方法里面也存在很多可以配合的参数，比如：获取刷盘间隔时间默认200ms（commitIntervalCommitLog进行配置），刷盘最少页码默认4即16k（commitCommitLogLeastPages进行配置），最长刷盘延迟时间默认200ms（commitCommitLogThoroughInterval进行配置）。

如果当前时间距离上次刷盘时间大于等于200ms，那么必定刷盘，因此设置刷盘的最少页数为0，更新刷盘时间戳为当前时间。

其它的整体和前面的类似，强调一点：堆内存直接进行flush而堆外内存则是提交到操作系统的PageCache，刷盘由操作系统控制。

主从同步

当我们搭建集群的时候，默认都是给Master节点发送数据，但是主节点和从节点之间也是需要进行数据的传输，RocketMQ在启动的时候也进行相关启动，等刷盘完成之后就会有一个通知主从同步的方法submitReplicaRequest()。

public CompletableFuture<PutMessageStatus> submitReplicaRequest(AppendMessageResult result, MessageExt messageExt) {
        
        if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
        
            // 构建HAServer对象，执行相关逻辑
            HAService service = this.defaultMessageStore.getHaService();
            if (messageExt.isWaitStoreMsgOK()) {
        
                if (service.isSlaveOK(result.getWroteBytes() + result.getWroteOffset())) {
        
                    GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                            this.defaultMessageStore.getMessageStoreConfig().getSlaveTimeout());
                    service.putRequest(request);
                    service.getWaitNotifyObject().wakeupAll();
                    return request.future();
                }
                else {
        
                    return CompletableFuture.completedFuture(PutMessageStatus.SLAVE_NOT_AVAILABLE);
                }
            }
        }
        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
    }

通过前面启动Broker执行HAService的start方法会进行初始化请求和启动三个服务。

public void start() throws Exception {
        
        this.acceptSocketService.beginAccept();
        // 维护master与slave端之间的tcp连接
        this.acceptSocketService.start();
        // 主从消息同步
        this.groupTransferService.start();
        // slave的客户端
        this.haClient.start();
    }

首先初始化请求，这里极大的追求性能的极致，在此处使用NIO进行传输而不是使用Netty，原因在于主从同步数据量大不便于使用更轻量级的Netty构建。

acceptSocketService的run方法用来维护Master与slave之间的连接，每隔1s中处理一次连接就是事件构建服务端数据传输通道创建HAConnection对象负责同步逻辑。

[c79e3c6ac47c4d80939c6e9030d715bd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/05/942f048ce03341b0b56fd64bf1dcaf60.png