成人午夜视频全免费观看高清-秋霞福利视频一区二区三区-国产精品久久久久电影小说-亚洲不卡区三一区三区一区

RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼

這篇文章主要介紹“RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼”，在日常操作中，相信很多人在RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

我們提供的服務(wù)有：網(wǎng)站制作、成都網(wǎng)站設(shè)計、微信公眾號開發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認證、臨城ssl等。為上1000家企事業(yè)單位解決了網(wǎng)站和推廣的問題。提供周到的售前咨詢和貼心的售后服務(wù)，是有科學(xué)管理、有技術(shù)的臨城網(wǎng)站制作公司

1、刷盤策略

RocketMQ提供了兩種刷盤策略同步刷盤、異步刷盤

同步刷盤：在消息到達MQ后，RocketMQ需要將數(shù)據(jù)持久化，同步刷盤是指數(shù)據(jù)到達內(nèi)存之后，必須刷到commitlog日志之后才算成功，然后返回producer數(shù)據(jù)已經(jīng)發(fā)送成功。

異步刷盤：，同步刷盤是指數(shù)據(jù)到達內(nèi)存之后,返回producer說數(shù)據(jù)已經(jīng)發(fā)送成功。，然后再寫入commitlog日志。

復(fù)制方式優(yōu)點缺點適應(yīng)場景
同步刷盤保證了消息不丟失吞吐率相對于異步刷盤要低消息可靠性要求較高的場景
異步刷盤系統(tǒng)的吞吐量提高系統(tǒng)斷電等異常時會有部分丟失對應(yīng)吞吐量要求較高的場景

下面我們從源碼的角度分析其實現(xiàn)的邏輯

RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼

2、同步刷盤

CommitLog.putMessage()方法中的刷盤的核心方法handleDiskFlush()

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
    // Synchronization flush  同步刷盤
    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
        //客戶端確認要等待刷盤成功
        if (messageExt.isWaitStoreMsgOK()) {
        	//封裝刷盤請求對象 nextoffset : 當(dāng)前內(nèi)存寫的位置 + 本次要寫入的字節(jié)數(shù)
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
            //添加刷盤請求（后臺定時任務(wù)進行刷盤，每隔10毫秒批量刷盤。10毫秒中如果有多個請求，則多個請求一塊刷盤）
            service.putRequest(request);
            //等待刷盤請求結(jié)果（最長等待5秒鐘，刷盤成功后馬上可以獲取結(jié)果。）
            boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
            if (!flushOK) {
                log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()
                    + " client address: " + messageExt.getBornHostString());
                putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
            }
        } else {
            service.wakeup();
        }
    }else {// Asynchronous flush 異步刷盤
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
        	//喚醒FlushRealTimeService服務(wù)線程
            flushCommitLogService.wakeup();
        } else {
        	//喚醒CommitRealTimeService服務(wù)線程
            commitLogService.wakeup();
        }
    }
}

查看同步刷盤的核心類GroupCommitService中的核心屬性

private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>(); private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>(); requestsWrite : 寫隊列，主要用于向該線程添加刷盤任務(wù) requestsRead : 讀隊列，主要用于執(zhí)行特定的刷盤任務(wù)，這是是GroupCommitService 設(shè)計的一個亮點，把讀寫分離，每處理完requestsRead中的任務(wù)，就交換這兩個隊列。

我們查看其run()方法

public void run() {
    CommitLog.log.info(this.getServiceName() + " service started");
    while (!this.isStopped()) {
        try {
        	//等待通知，如果數(shù)據(jù)過來，提前結(jié)束等待執(zhí)行onWaitEnd()方法交換讀寫swapRequests（）
        	//刷盤請求的requestsWrite->requestsRead
            this.waitForRunning(10);
            //執(zhí)行刷盤
            this.doCommit();
        } catch (Exception e) {
            CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }
    //省略代碼...
}

waitForRunning方法中執(zhí)行了swapRequests()方法

private void swapRequests() {
    List<GroupCommitRequest> tmp = this.requestsWrite;
    this.requestsWrite = this.requestsRead;
    this.requestsRead = tmp;
}

GroupCommitService接收到的刷盤請求通過putRequest()方法加入到requestsWrite集合中，swapRequests()方法將requestsWrite請求集合交換到requestsRead集合中供刷盤使用，我們重點查看doCommit()方法

private void doCommit() {
    synchronized (this.requestsRead) {
        if (!this.requestsRead.isEmpty()) {
        	//循環(huán)每一個刷盤請求
            for (GroupCommitRequest req : this.requestsRead) {
                // There may be a message in the next file, so a maximum of
                // two times the flush
                boolean flushOK = false;
                for (int i = 0; i < 2 && !flushOK; i++) {
                	//判斷是否已經(jīng)刷盤過了，刷盤的位置和當(dāng)前消息下次刷盤需要的位置比較
                    flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                    if (!flushOK) {
                    	//0代碼立刻刷盤，不管緩存中消息有多少
                        CommitLog.this.mappedFileQueue.flush(0);
                    }
                }
                //返回刷盤的結(jié)果
                req.wakeupCustomer(flushOK);
            }
            long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
            //設(shè)置刷盤的時間點
            if (storeTimestamp > 0) {
                CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
            }
            //清空requestsRead對象
            this.requestsRead.clear();
        } else {
            // Because of individual messages is set to not sync flush, it
            // will come to this process
            CommitLog.this.mappedFileQueue.flush(0);
        }
    }
}

mappedFileQueue.flush(0)立刻刷盤

public boolean flush(final int flushLeastPages) {
    boolean result = true;
    MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);
    if (mappedFile != null) {
        long tmpTimeStamp = mappedFile.getStoreTimestamp();
        //刷盤，返回刷寫到磁盤指針
        int offset = mappedFile.flush(flushLeastPages);
        //計算當(dāng)前的刷盤指針，之前的所有數(shù)據(jù)已經(jīng)持久化到磁盤中
        long where = mappedFile.getFileFromOffset() + offset;
        result = where == this.flushedWhere;
        this.flushedWhere = where;
        if (0 == flushLeastPages) {
            this.storeTimestamp = tmpTimeStamp;
        }
    }
    return result;
}

mappedFile.flush(0);保證立刻刷盤后面異步刷盤時也會調(diào)用mappedFile.flush()方法

3、異步刷盤

if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
	//喚醒FlushRealTimeService服務(wù)線程
    flushCommitLogService.wakeup();
} else {
	//喚醒CommitRealTimeService服務(wù)線程
    commitLogService.wakeup();
}

我們發(fā)現(xiàn)異步刷盤的時候有兩種方式，一種是堆外內(nèi)存池開啟時啟動CommitRealTimeService服務(wù)線程，另一個是默認執(zhí)行的FlushRealTimeService服務(wù)線程進行刷盤操作，關(guān)于TransientStorePoolEnable在《RocketMQ內(nèi)存映射》章節(jié)中的**“創(chuàng)建映射文件MappedFile”**中有介紹

RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼

圖3-1

1、FlushRealTimeService

查看其run()方法

public void run() {
    CommitLog.log.info(this.getServiceName() + " service started");
    while (!this.isStopped()) {
    	// 每次刷盤的間隔時間，默認 200ms
        int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
        // 每次commit最少的頁數(shù) 默認4頁
        int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
        // 如果上次刷新的時間+該值 小于當(dāng)前時間，則改變flushPhysicQueueLeastPages =0 默認為200
        int commitDataThoroughInterval =
            CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();

        long begin = System.currentTimeMillis();
        //距離上一次刷盤時間超過200ms則立刻刷盤，commit最少的頁數(shù)置為0
        if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
            this.lastCommitTimestamp = begin;
            commitDataLeastPages = 0;
        }
        try {
        	//刷盤
            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;
    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");
    }
}

這種方式和同步刷盤一樣就是mappedFileQueue.commit(commitDataLeastPages)參數(shù)有限制，數(shù)據(jù)達到一定量的時候才進行刷盤操作提高數(shù)據(jù)的刷盤性能。

2、CommitRealTimeService

查看其run()方法

public void run() {
    CommitLog.log.info(this.getServiceName() + " service started");
    while (!this.isStopped()) {
    	// 每次刷盤的間隔時間，默認 200ms
        int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
        // 每次commit最少的頁數(shù) 默認4頁
        int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
        // 如果上次刷新的時間+該值 小于當(dāng)前時間，則改變flushPhysicQueueLeastPages =0 默認為200
        int commitDataThoroughInterval =
            CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();

        long begin = System.currentTimeMillis();
        //距離上一次刷盤時間超過200ms則立刻刷盤，commit最少的頁數(shù)置為0
        if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
            this.lastCommitTimestamp = begin;
            commitDataLeastPages = 0;
        }
        try {
        	//刷盤
            boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
            long end = System.currentTimeMillis();
            //返回的是false說明數(shù)據(jù)已經(jīng)commit到了fileChannel中
            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;
    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");
    }
}

我們發(fā)現(xiàn)其刷盤方法不一樣mappedFileQueue.commit()調(diào)用MappedFile.commit()方法

public int commit(final int commitLeastPages) {
    if (writeBuffer == null) {
        //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
        return this.wrotePosition.get();
    }
    //如果提交的數(shù)據(jù)不滿commitLeastPages則不執(zhí)行本次的提交，待下一次提交
    if (this.isAbleToCommit(commitLeastPages)) {
        if (this.hold()) {
            commit0(commitLeastPages);
            this.release();
        } else {
            log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
        }
    }

    // All dirty data has been committed to FileChannel.
    if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
        this.transientStorePool.returnBuffer(writeBuffer);
        this.writeBuffer = null;
    }

    return this.committedPosition.get();
}

查看其核心刷盤方法

protected void commit0(final int commitLeastPages) {
    int writePos = this.wrotePosition.get();
    int lastCommittedPosition = this.committedPosition.get();
    if (writePos - this.committedPosition.get() > 0) {
        try {
        	//創(chuàng)建writeBuffer的共享緩存區(qū)
            ByteBuffer byteBuffer = writeBuffer.slice();
            //將指針回退到上一次提交的位置
            byteBuffer.position(lastCommittedPosition);
            //設(shè)置limit為writePos
            byteBuffer.limit(writePos);
            this.fileChannel.position(lastCommittedPosition);
            //將committedPosition指針到wrotePosition的數(shù)據(jù)復(fù)制（寫入）到fileChannel中
            this.fileChannel.write(byteBuffer);
            //更新committedPosition指針為writePos
            this.committedPosition.set(writePos);
        } catch (Throwable e) {
            log.error("Error occurred when commit data to FileChannel.", e);
        }
    }
}

commit0（）只是將緩存數(shù)據(jù)加入到fileChannel中，我們在CommitRealTimeService.run()方法中看到喚醒flushCommitLogService線程需要將fileChannel中的數(shù)據(jù)flush到磁盤中，我們發(fā)現(xiàn)兩種方式都需要走flushCommitLogService.run()方法最后都執(zhí)行MappedFile.flush(int)

public int flush(final int flushLeastPages) {
    if (this.isAbleToFlush(flushLeastPages)) {
        if (this.hold()) {
            int value = getReadPosition();
            try {
                //We only append data to fileChannel or mappedByteBuffer, never both.
                if (writeBuffer != null || this.fileChannel.position() != 0) {
                    this.fileChannel.force(false);
                } else {
                    this.mappedByteBuffer.force();
                }
            } catch (Throwable e) {
                log.error("Error occurred when force data to disk.", e);
            }
            //設(shè)置刷盤后的指針
            this.flushedPosition.set(value);
            this.release();
        } else {
            log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());
            this.flushedPosition.set(getReadPosition());
        }
    }
    return this.getFlushedPosition();
}

兩種緩存方式走的刷盤邏輯也不同，可以查看**“圖3-1”**兩種方式的處理流程圖

我們還發(fā)現(xiàn)一個方法isAbleToFlush()判斷是否需要刷盤

private boolean isAbleToFlush(final int flushLeastPages) {
    int flush = this.flushedPosition.get();
    int write = getReadPosition();
    if (this.isFull()) {
        return true;
    }
    if (flushLeastPages > 0) {
        return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
    }
    return write > flush;
}

同步刷盤時flushLeastPages=0立刻刷盤

異步刷盤時flushLeastPages=4 ,默認是4，需要刷盤的數(shù)據(jù)達到PageCache的頁數(shù)4倍時才會刷盤，或者距上一次刷盤時間>=200ms則設(shè)置flushLeastPages=0立刻刷盤

同步刷盤時無論消息的大小都立刻刷盤，線程阻塞等待刷盤結(jié)果

異步刷盤有兩種方式但是其邏輯都是需要刷盤的數(shù)據(jù)OS_PAGE_SIZE的4倍即（1024 * 4）*4=16k或者距上一次刷盤時間>=200ms時才刷盤，提高數(shù)據(jù)的刷盤性能

到此，關(guān)于“RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>

當(dāng)前標(biāo)題：RocketMQ的刷盤策略以及實現(xiàn)同步刷盤和異步刷盤的實例代碼
URL分享：http://jinyejixie.com/article16/jjpidg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供定制開發(fā)、ChatGPT、面包屑導(dǎo)航、移動網(wǎng)站建設(shè)、、企業(yè)網(wǎng)站制作

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)

外貿(mào)網(wǎng)站建設(shè)
兴文县| 金山区| 竹北市| 三河市| 大足县| 重庆市| 浏阳市| 博湖县| 蒙自县| 九龙坡区| 太仆寺旗| 西乌珠穆沁旗| 界首市| 龙门县| 金湖县| 泰宁县| 台南市| 辉南县| 定日县| 云和县| 平谷区| 临朐县| 怀集县| 亚东县| 海淀区| 密山市| 永春县| 应城市| 名山县| 泸水县| 广南县| 凯里市| 牟定县| 东莞市| 永春县| 铜鼓县| 辽源市| 民丰县| 儋州市| 南郑县| 郎溪县|