前文目錄鏈接參考：

 消息隊列的一些場景及源碼分析，RocketMQ使用相關問題及性能優化 //www.cnblogs.com/yizhiamumu/p/16694126.html

消息隊列的對比測試與RocketMQ使用擴展 //www.cnblogs.com/yizhiamumu/p/16677881.html

消息隊列為什麼選用redis？聊聊如何做技術方案選型？//www.cnblogs.com/yizhiamumu/p/16573472.html

分佈式事務原理及解決方案案例 //www.cnblogs.com/yizhiamumu/p/16662412.html

分佈式事務實戰方案匯總 //www.cnblogs.com/yizhiamumu/p/16625677.html

消息隊列初見：一起聊聊引入系統mq 之後的問題 //www.cnblogs.com/yizhiamumu/p/16573472.html

參考：消息隊列為什麼選用redis？聊聊如何做技術方案選型？//www.cnblogs.com/yizhiamumu/p/16573472.html

上文，我們把 Redis 當作隊列來使用時，始終面臨的 2 個問題：

1. Redis 本身可能會丟數據
2. 面對消息積壓，Redis 內存資源緊張

如果你的業務場景足夠簡單，對於數據丟失不敏感，而且消息積壓概率比較小的情況下，把 Redis 當作隊列是完全可以的。

 一：消息隊列的一些場景

1.1 為什麼有各種各樣的 MQ？

近幾年，確實出現了很多消息隊列解決方案，但其實去分析每種消息隊列，會發現他們誕生的背景和要針對性解決的問題是不一樣的。

• RabbitMQ 誕生於標準化與開源，打破了商業化消息隊列的技術壁壘，但應用場景其實沒變，定位為異步與解耦；
• Kafka 誕生的背景是大數據，以批量，高吞吐等核心能力搶佔了大數據管道的心智，隨後非常自然地定位到 Streaming 領域；
• EMQ 重點聚焦的領域在物聯網，物聯網的挑戰跟其他領域是大相徑庭的，超大規模的設備與連接數，規則引擎，甚者邊緣段需要有一整套完整的解決方案；
• Pulsar 作為後起之秀嘗試在多個領域發力，包括 Messaging、Function、Streaming 等多領域都有相應布局。

回到 RocketMQ，大家能從近兩年 RocketMQ 在社區的一系列動作中發現，RocketMQ 同時在消息、事件、流三個領域都有發力，逐漸演進至一個超融合處理平台。作為一個融合的數據處理平台，RocketMQ 當前在開源的布局看起來是與業界多個 MQ 趨同，在 RocketMQ 開源的背後其實是商業上真實的需求驅動。

1.2 從性能上來講，相關基準測試數據是什麼水平？

一般講性能，其實就是吞吐量和延遲兩個指標。

對於吞吐量來講，RocketMQ 在 2017 年就能優化到單機 50W 的 TPS。如果是在批量的場景，實際上從生產環境的穩定性，以及業務消息的重要性來講，各個消息隊列都能輕易地打滿網絡帶寬或者磁盤資源。

也就是說，性能一般情況下差異都不大，是很難作為一個產品的核心競爭力的，除非是架構層面有限制。

延遲就是一個非常重要的指標了，在線業務對於是 2ms 延時和 5ms 延時基本上都能接受，但非常難以接受的是經常性有秒級的毛刺（在延遲這個指標後面長尾延遲）。

除了上述兩點，彈性和可擴展能力也是非常重要的。

1.3 消息如何存儲

消息我們可以直接在內存中使用數組或者隊列來存儲數據即可。性能非常高。

但是有幾方面的缺點

1. 數據丟失，比如異常情況服務器宕機重啟後內存的消息會被丟失掉
2. 數據量大的時候，內存放不下，或者需要高昂的成本。如果 面對一些業務系統是不能容忍消息丟失的情況，單純放內存存儲也不太可能，所以需要一款持久化的消息系統。

既然要存儲數據，就需要解決數據存哪裡？從存儲方式來看，主要有幾個方面：

• 關係型數據庫，比如mysql
• 分佈式KV存儲，比如採用rocketdb實現的
• 文件系統，log 的方式直接追加

性能，吞吐量，本質上就是數據結構的設計決定的。我們看看上面數據存儲方式對應的數據結構

mysql 在大數據量的情況，性能會急劇下降，並且擴展性非常不友好。

分佈式KV 存儲 天然的分佈式系統，對大數據量和未來的擴展都問題不大，LSM tree 對寫性能和吞吐都比mysql 要好。查詢其實是可以通過緩存等手段去優化，可以考慮。

但是，滿足性能和吞吐量最優的毫無疑問是使用文件系統，因為消息不需要修改，讀和寫都是順序讀寫，性能極高。

但是現實中的需求我們可能需要使用多個隊列來完成不同的業務。比如一個隊列來處理訂單相關的業務，一個隊列來處理商品相關的業務等等。那麼我們該如何調整呢？

我們都知道文件 append only log 的方式是不支持根據消息的內容來搜索的，如果所有的隊列的數據存在一個文件中，是沒辦法滿足需求的。

換個思路，一個隊列一個文件我們就可以繞開根據內容檢索的需求，kafka 就是這麼玩的。

這個時候，每個隊列一個文件，讀寫還是順序的嗎？

我們現在面臨的問題是，作為一款面向業務的高性能消息中間件，隨着業務的複雜度變高，隊列數量是急劇變大的。

如果要保證寫入的吞吐量和性能，還需要得所有的隊列都寫在同一個文件。

但是，按照隊列消費的場景就意味着要根據消息內容（隊列名字）來進行消費，append only log 是不支持檢索的，如何解決這個問題。

我們會增加一個索引來處理慢sql 。我們是否也可以建立一個隊列的索引，每一個隊列就是一個索引文件。

讀取數據的時候，先從索引隊列找到消息在文件的偏移量後，在到數據文件去讀取。

那麼，索引的文件的數量變大的之後，那麼對索引文件的讀寫不就是又變成隨機讀寫了嗎？性能又會急劇下降？

一個一個來解決：

1. 寫索引文件的時候，我們可以改成異步寫，也就是寫完數據文件，可以直接返回給客戶端成功了，後台再由一個線程不停的從數據文件獲取數據來構建索引，這樣就可以解決寫的性能瓶頸了
2. 讀的問題，我們要盡量避免直接從磁盤讀，改成從內存讀。放在內存就意味着索引的內容要足夠小，不然根本放不下。所以問題就變成盡量控制索引文件的大小，放在內存裏面來避開磁盤讀從而提高性能

 （rocketmq 中數據文件稱為：commitlog,   topic索引文件稱為 consumeQueue）

我們得出結論：選擇文件系統，append only log.根據消息隊列即時消費和順序讀寫的特點，剛寫入的內容還在page cache ,就被讀走了，甚至都不需要回到磁盤，性能會非常高。

1.5 數據量大了存儲怎麼辦

本地切割，大文件變小文件

如果所有的數據都存在一個commitlog 文件的話，隨着數據量變大，文件必然會非常大。

解決方案是，我們大文件切換成小文件，每個文件固定大小1G，寫滿了就切換到一個新的文件

分佈式存儲

消息隊列的第一個特點就是數據量大，一台機器容易面臨瓶頸，因此我們需要把數據均衡的分發到各個機器上。

解決方案是，一段很長的隊列平均切成N份，把這N份分別放到不同的機器上

1.6 消息高可靠

雖然消息已經分成切分成為多份放到不同的機器了，但是每一份都是都只有一個副本，也就意味着，任何一台機器的硬盤壞掉的話，該機器上的消息就會丟失掉了，這是不可接受的。

行業通常的做法一份數據存多個副本，並且確保所有的副本不能全都在同一台機器。

問題來了，那麼這多份數據是同步雙寫還是異步雙寫呢？

其實每個業務場景需求是不一樣的，RocketMq 是支持可配置的

1.7 Broker是怎麼保存數據的呢？

RocketMQ主要的存儲文件包括CommitLog文件、ConsumeQueue文件、Indexfile文件

所以，Broker是怎麼保存數據的呢？Broker在收到消息之後，會把消息保存到commitlog的文件當中，而同時在分佈式的存儲當中，每個broker都會保存一部分topic的數據，同時，每個topic對應的messagequeue下都會生成consumequeue文件用於保存commitlog的物理位置偏移量offset，indexfile中會保存key和offset的對應關係。

 

Broker

CommitLog文件保存於${Rocket_Home}/store/commitlog目錄中，從圖中我們可以明顯看出來文件名的偏移量，每個文件默認1G，寫滿後自動生成一個新的文件。

 

log

 

由於同一個topic的消息並不是連續的存儲在commitlog中，消費者如果直接從commitlog獲取消息效率非常低，所以通過consumequeue保存commitlog中消息的偏移量的物理地址，這樣消費者在消費的時候先從consumequeue中根據偏移量定位到具體的commitlog物理文件，然後根據一定的規則（offset和文件大小取模）在commitlog中快速定位。

 

log

1.8 RocketMQ怎麼對文件進行讀寫

RocketMQ對文件的讀寫巧妙地利用了操作系統的一些高效文件讀寫方式——PageCache、順序讀寫、零拷貝

1.8.1 PageCache、順序讀取

在RocketMQ中，ConsumeQueue邏輯消費隊列存儲的數據較少，並且是順序讀取，在page cache機制的預讀取作用下，Consume Queue文件的讀性能幾乎接近讀內存，即使在有消息堆積情況下也不會影響性能。而對於CommitLog消息存儲的日誌數據文件來說，讀取消息內容時候會產生較多的隨機訪問讀取，嚴重影響性能。如果選擇合適的系統IO調度算法，比如設置調度算法為Deadline（此時塊存儲採用SSD的話），隨機讀的性能也會有所提升。

頁緩存（PageCache)是OS對文件的緩存，用於加速對文件的讀寫。一般來說，程序對文件進行順序讀寫的速度幾乎接近於內存的讀寫速度，主要原因就是由於OS使用PageCache機制對讀寫訪問操作進行了性能優化，將一部分的內存用作PageCache。對於數據的寫入，OS會先寫入至Cache內，隨後通過異步的方式由pdflush內核線程將Cache內的數據刷盤至物理磁盤上。對於數據的讀取，如果一次讀取文件時出現未命中PageCache的情況，OS從物理磁盤上訪問讀取文件的同時，會順序對其他相鄰塊的數據文件進行預讀取

1.8.2 零拷貝

RocketMQ主要通過MappedByteBuffer對文件進行讀寫操作。其中，利用了NIO中的FileChannel模型將磁盤上的物理文件直接映射到用戶態的內存地址中（這種Mmap的方式減少了傳統IO，將磁盤文件數據在操作系統內核地址空間的緩衝區，和用戶應用程序地址空間的緩衝區之間來回進行拷貝的性能開銷），將對文件的操作轉化為直接對內存地址進行操作，從而極大地提高了文件的讀寫效率（正因為需要使用內存映射機制，故RocketMQ的文件存儲都使用定長結構來存儲，方便一次將整個文件映射至內存）。

什麼是零拷貝
在操作系統中，使用傳統的方式，數據需要經歷幾次拷貝，還要經歷用戶態/內核態切換

1. 從磁盤複製數據到內核態內存；
2. 從內核態內存複製到用戶態內存；
3. 然後從用戶態內存複製到網絡驅動的內核態內存；
4. 最後是從網絡驅動的內核態內存複製到網卡中進行傳輸。

 

 

 傳統文件傳輸示意圖

 

所以，可以通過零拷貝的方式，減少用戶態與內核態的上下文切換和內存拷貝的次數，用來提升I/O的性能。零拷貝比較常見的實現方式是mmap，這種機制在Java中是通過MappedByteBuffer實現的。

 

 

 mmap示意圖

 

 

1.9 消息刷盤怎麼實現

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

• 同步刷盤：在消息達到Broker的內存之後，必須刷到commitLog日誌文件中才算成功，然後返回Producer數據已經發送成功。
• 異步刷盤：異步刷盤是指消息達到Broker內存後就返回Producer數據已經發送成功，會喚醒一個線程去將數據持久化到CommitLog日誌文件中

Broker在消息的存取時直接操作的是內存（內存映射文件），這可以提供系統的吞吐量，但是無法避免機器掉電時數據丟失，所以需要持久化到磁盤中

刷盤的最終實現都是使用NIO中的 MappedByteBuffer.force() 將映射區的數據寫入到磁盤，如果是同步刷盤的話，在Broker把消息寫到CommitLog映射區後，就會等待寫入完成

異步而言，只是喚醒對應的線程，不保證執行的時機，

Producer負載均衡：索引遞增隨機取模
public MessageQueue selectOneMessageQueue(){
    //索引遞增
    int index = this.sendWhichQueue.incrementAndGet();
    //利用索引取隨機數，取餘數
    int pos = Math.abs(index) % this.messageQueueList.size();
    if(pos<0){
        pos=0;  
    }
    return this.messageQueueList.get(pos);
}

 //如果沒有拉到數據
case ResponseCode.PULL_NOT_FOUND:
// broker 和 consumer 都允許 suspend，默認開啟
if (brokerAllowSuspend && hasSuspendFlag) {
    long pollingTimeMills = suspendTimeoutMillisLong;
    if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
          pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
     }

    String topic = requestHeader.getTopic();
    long offset = requestHeader.getQueueOffset();
    int queueId = requestHeader.getQueueId();
    //封裝一個PullRequest
    PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
    this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
   //把PullRequest掛起來
   this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
     response = null;
     break;
}

@Override
public void run() {
   log.info("{} service started", this.getServiceName());
   while (!this.isStopped()) {
       try {
          if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                 this.waitForRunning(5 * 1000);
          } else {
                    this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                }

         long beginLockTimestamp = this.systemClock.now();
         //檢查hold住的請求
         this.checkHoldRequest();
         long costTime = this.systemClock.now() - beginLockTimestamp;
         if (costTime > 5 * 1000) {
              log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
         }
        } catch (Throwable e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }
    log.info("{} service end", this.getServiceName());
    }

1.12 RocketMQ為什麼速度快？

是因為使用了順序存儲、Page Cache和異步刷盤。

1、我們在寫入commitlog的時候是順序寫入的，這樣比隨機寫入的性能就會提高很多。

2、寫入commitlog的時候並不是直接寫入磁盤，而是先寫入操作系統的PageCache。

3、最後由操作系統異步將緩存中的數據刷到磁盤。

二：RocketMQ的基本架構

 RocketMQ 一共有四個部分組成：NameServer，Broker，Producer 生產者，Consumer 消費者，它們對應了：發現、發、存、收，為了保證高可用，一般每一部分都是集群部署的

1.1 NameServer

NameServer 是一個無狀態的服務器，角色類似於 Kafka使用的 Zookeeper，但比 Zookeeper 更輕量。

特點：
每個 NameServer 結點之間是相互獨立，彼此沒有任何信息交互。
Nameserver 被設計成幾乎是無狀態的，通過部署多個結點來標識自己是一個偽集群，Producer 在發送消息前從 NameServer中獲取 Topic 的路由信息也就是發往哪個 Broker，Consumer 也會定時從 NameServer獲取 Topic的路由信息，Broker 在啟動時會向 NameServer 註冊，並定時進行心跳連接，且定時同步維護的 Topic 到 NameServer
功能主要有兩個：

• 和Broker 結點保持長連接。
• 維護 Topic 的路由信息。

1.2 Broker

消息存儲和中轉角色，負責存儲和轉發消息

Broker 內部維護着一個個 Consumer Queue，用來存儲消息的索引，真正存儲消息的地方是 CommitLog（日誌文件）

單個 Broker 與所有的 Nameserver 保持着長連接和心跳，並會定時將 Topic 信息同步到 NameServer，和 NameServer 的通信底層是通過 Netty 實現的。

1.3 Producer

消息生產者，業務端負責發送消息，由用戶自行實現和分佈式部署。

Producer由用戶進行分佈式部署，消息由Producer通過多種負載均衡模式發送到Broker集群，發送低延時，支持快速失敗。
RocketMQ 提供了三種方式發送消息：同步、異步和單向

• 同步發送：同步發送指消息發送方發出數據後會在收到接收方發迴響應之後才發下一個數據包。一般用於重要通知消息，例如重要通知郵件、營銷短訊。
• 異步發送：異步發送指發送方發出數據後，不等接收方發迴響應，接着發送下個數據包，一般用於可能鏈路耗時較長而對響應時間敏感的業務場景，例如用戶視頻上傳後通知啟動轉碼服務。
• 單向發送：單向發送是指只負責發送消息而不等待服務器回應且沒有回調函數觸發，適用於某些耗時非常短但對可靠性要求並不高的場景，例如日誌收集

1.4 Consumer

消息消費者，負責消費消息，一般是後台系統負責異步消費。

Consumer也由用戶部署，支持PUSH和PULL兩種消費模式，支持集群消費和廣播消費，提供實時的消息訂閱機制。

• Pull：拉取型消費者（Pull Consumer）主動從消息服務器拉取信息，只要批量拉取到消息，用戶應用就會啟動消費過程，所以 Pull 稱為主動消費型
• Push：推送型消費者（Push Consumer）封裝了消息的拉取、消費進度和其他的內部維護工作，將消息到達時執行的回調接口留給用戶應用程序來實現。所以 Push 稱為被動消費類型，但其實從實現上看還是從消息服務器中拉取消息，不同於 Pull 的是 Push 首先要註冊消費監聽器，當監聽器處觸發後才開始消費消息

2 RocketMQ 原理

2.1 RocketMQ整體工作流程

RocketMQ是一個分佈式消息隊列，也就是消息隊列+分佈式系統

作為消息隊列，它是發-存-收的一個模型，對應的就是Producer、Broker、Cosumer；作為分佈式系統，它要有服務端、客戶端、註冊中心，對應的就是Broker、Producer/Consumer、NameServer

主要的工作流程：RocketMQ由NameServer註冊中心集群、Producer生產者集群、Consumer消費者集群和若干Broker（RocketMQ進程）組成：

1. Broker在啟動的時候去向所有的NameServer註冊，並保持長連接，每30s發送一次心跳
2. Producer在發送消息的時候從NameServer獲取Broker服務器地址，根據負載均衡算法選擇一台服務器來發送消息
3. Conusmer消費消息的時候同樣從NameServer獲取Broker地址，然後主動拉取消息來消費

2.2 如何保證RocketMQ的高可用？

NameServer因為是無狀態，且不相互通信的，所以只要集群部署就可以保證高可用。

RocketMQ的高可用主要是在體現在Broker的讀和寫的高可用，Broker的高可用是通過集群和主從實現的。

Broker可以配置兩種角色：Master和Slave，Master角色的Broker支持讀和寫，Slave角色的Broker只支持讀，Master會向Slave同步消息。

也就是說Producer只能向Master角色的Broker寫入消息，Cosumer可以從Master和Slave角色的Broker讀取消息。

Consumer 的配置文件中，並不需要設置是從 Master 讀還是從 Slave讀，當 Master 不可用或者繁忙的時候， Consumer 的讀請求會被自動切換到從 Slave。有了自動切換 Consumer 這種機制，當一個 Master 角色的機器出現故障後，Consumer 仍然可以從 Slave 讀取消息，不影響 Consumer 讀取消息，這就實現了讀的高可用。

如何達到發送端寫的高可用性呢？

在創建 Topic 的時候，把 Topic 的多個Message Queue 創建在多個 Broker 組上（相同 Broker 名稱，不同 brokerId機器組成 Broker 組），這樣當 Broker 組的 Master 不可用後，其他組Master 仍然可用， Producer 仍然可以發送消息 RocketMQ 目前還不支持把Slave自動轉成 Master ，如果機器資源不足，需要把 Slave 轉成 Master ，則要手動停止 Slave 色的 Broker ，更改配置文件，用新的配置文件啟動 Broker。

2.3 Master和Slave之間是怎麼同步數據的呢？

而消息在master和slave之間的同步是根據raft協議來進行的：

1、在broker收到消息後，會被標記為uncommitted狀態
2、然後會把消息發送給所有的slave
3、slave在收到消息之後返回ack響應給master
4、master在收到超過半數的ack之後，把消息標記為committed
5、發送committed消息給所有slave，slave也修改狀態為committed

2.4 為什麼RocketMQ不使用Zookeeper作為註冊中心呢？

Kafka採用Zookeeper作為註冊中心（也開始逐漸去Zookeeper），

RocketMQ不使用Zookeeper其實主要可能從這幾方面來考慮：

1. 基於可用性的考慮，根據CAP理論【Consistency(一致性)、Availability(可用性)、Partition Tolerance(分區容錯性)，不能同時成立】，Zookeeper滿足的是CP，並不能保證服務的可用性，Zookeeper在進行選舉的時候，整個選舉的時間太長，期間整個集群都處於不可用的狀態，而這對於一個註冊中心來說肯定是不能接受的，作為服務發現來說就應該是為可用性而設計。
2. 基於性能的考慮，NameServer本身的實現非常輕量，而且可以通過增加機器的方式水平擴展，增加集群的抗壓能力，而Zookeeper的寫是不可擴展的，Zookeeper要解決這個問題只能通過劃分領域，劃分多個Zookeeper集群來解決，首先操作起來太複雜，其次這樣還是又違反了CAP中的A的設計，導致服務之間是不連通的。
3. 持久化的機制來帶的問題，ZooKeeper 的 ZAB 協議對每一個寫請求，會在每個 ZooKeeper  節點上保持寫一個事務日誌，同時再加上定期的將內存數據鏡像（Snapshot）到磁盤來保證數據的一致性和持久性，而對於一個簡單的服務發現的場景來說，這其實沒有太大的必要，這個實現方案太重了。而且本身存儲的數據應該是高度定製化的。
4. 消息發送應該弱依賴註冊中心，這也是RocketMQ的設計理念，生產者在第一次發送消息的時候從NameServer獲取到Broker地址後緩存到本地，如果NameServer整個集群不可用，短時間內對於生產者和消費者並不會產生太大影響。

 三：RocketMQ使用相關問題

1. 如何保證消息的可用性/可靠性/不丟失呢？

消息的一個處理方式是異步發送，那消息可靠性怎麼保證？

消息丟失可能發生在生產者發送消息、MQ本身丟失消息、消費者丟失消息3個方面。

1.1 生產者丟失

生產者丟失消息的可能點在於程序發送失敗拋異常了沒有重試處理，或者發送的過程成功但是過程中網絡閃斷MQ沒收到，消息就丟失了。

由於同步發送的一般不會出現這樣使用方式。

異步發送的場景下，一般分為兩個方式：異步有回調和異步無回調，無回調的方式，生產者發送完後不管結果可能就會造成消息丟失，而通過異步發送+回調通知+本地消息表的形式我們就可以做出一個解決方案。

所以在生產階段，主要通過請求確認機制，來保證消息的可靠傳遞。

• 1、同步發送的時候，要注意處理響應結果和異常。如果返迴響應OK，表示消息成功發送到了Broker，如果響應失敗，或者發生其它異常，都應該重試。
• 2、異步發送的時候，應該在回調方法里檢查，如果發送失敗或者異常，都應該進行重試。
• 3、如果發生超時的情況，也可以通過查詢日誌的API，來檢查是否在Broker存儲成功

以下單的場景舉例。

1、下單後先保存本地數據和MQ消息表，這時候消息的狀態是發送中，如果本地事務失敗，那麼下單失敗，事務回滾（訂單數據、MQ消息記錄都不會保存）。

2、下單成功，直接返回客戶端成功，異步發送MQ消息。

3、MQ回調通知消息發送結果，對應更新數據庫MQ發送狀態。

4、JOB輪詢超過一定時間（時間根據業務配置）還未發送成功的消息去重試
在監控平台配置或者JOB程序處理超過一定次數一直發送不成功的消息，告警，人工介入。

 

MQ

異步回調的形式是適合大部分場景下的一種解決方案。

1.2 MQ 存儲丟失

如果生產者保證消息發送到MQ，而MQ收到消息後還在內存中，這時候宕機了又沒來得及同步給從節點，就有可能導致消息丟失。

比如RocketMQ：

RocketMQ分為同步刷盤和異步刷盤兩種方式，默認的是異步刷盤，就有可能導致消息還未刷到硬盤上就丟失了，可以通過設置為同步刷盤的方式來保證消息可靠性，這樣即使MQ掛了，恢復的時候也可以從磁盤中去恢復消息。

比如Kafka也可以通過配置做到：

acks=all 只有參與複製的所有節點全部收到消息，才返回生產者成功。這樣的話除非所有的節點都掛了，消息才會丟失。
replication.factor=N,設置大於1的數，這會要求每個partion至少有2個副本
min.insync.replicas=N，設置大於1的數，這會要求leader至少感知到一個follower還保持着連接
retries=N，設置一個非常大的值，讓生產者發送失敗一直重試

雖然我們可以通過配置的方式來達到MQ本身高可用的目的，但是都對性能有損耗，怎樣配置需要根據業務做出權衡。

所以存儲階段，可以通過配置可靠性優先的 Broker 參數來避免因為宕機丟消息，簡單說就是可靠性優先的場景都應該使用同步。

• 1、消息只要持久化到CommitLog（日誌文件）中，即使Broker宕機，未消費的消息也能重新恢復再消費。
• 2、Broker的刷盤機制：同步刷盤和異步刷盤，不管哪種刷盤都可以保證消息一定存儲在pagecache中（內存中），但是同步刷盤更可靠，它是Producer發送消息後等數據持久化到磁盤之後再返迴響應給Producer。

• 3、Broker通過主從模式來保證高可用，Broker支持Master和Slave同步複製、Master和Slave異步複製模式，生產者的消息都是發送給Master，但是消費既可以從Master消費，也可以從Slave消費。同步複製模式可以保證即使Master宕機，消息肯定在Slave中有備份，保證了消息不會丟失。

 圖:同步刷盤和異步刷盤

1.3 消費者丟失

消費者丟失消息的場景1：消費者剛收到消息，此時服務器宕機，MQ認為消費者已經消費，不會重複發送消息，消息丟失。

RocketMQ默認是需要消費者回復ack確認，而kafka需要手動開啟配置關閉自動offset。

消費方不返回ack確認，重發的機制根據MQ類型的不同發送時間間隔、次數都不盡相同，如果重試超過次數之後會進入死信隊列，需要手工來處理了。（Kafka沒有這些）

 

MQ

消費者丟失消息的場景2：消費者收到消息，但消費業務邏輯出錯，消費失敗。
解決：利用前面提到的MQ本地表，消費者收到消息且業務邏輯執行完畢後再更新MQ消息的狀態（更新為已消費）

所以從Consumer角度分析，如何保證消息被成功消費？

• Consumer保證消息成功消費的關鍵在於確認的時機，不要在收到消息後就立即發送消費確認，而是應該在執行完所有消費業務邏輯之後，再發送消費確認。

因為消息隊列維護了消費的位置，邏輯執行失敗了，沒有確認，再去隊列拉取消息，就還是之前的一條。

2 如何處理消息重複的問題呢？

對分佈式消息隊列來說，同時做到確保一定投遞和不重複投遞是很難的，就是所謂的「有且僅有一次」 。RocketMQ擇了確保一定投遞，保證消息不丟失，但有可能造成消息重複。

處理消息重複問題，主要有業務端自己保證，主要的方式有兩種：業務冪等和消息去重。

業務冪等：第一種是保證消費邏輯的冪等性，也就是多次調用和一次調用的效果是一樣的。這樣一來，不管消息消費多少次，對業務都沒有影響。

消息去重：第二種是業務端，對重複的消息就不再消費了。這種方法，需要保證每條消息都有一個惟一的編號，通常是業務相關的，比如訂單號，消費的記錄需要落庫，而且需要保證和消息確認這一步的原子性。

具體做法是可以建立一個消費記錄表，拿到這個消息做數據庫的insert操作。給這個消息做一個唯一主鍵（primary key）或者唯一約束，那麼就算出現重複消費的情況，就會導致主鍵衝突，那麼就不再處理這條消息。

3 怎麼處理消息積壓？

3.1 消息積壓

發生了消息積壓，這時候就得想辦法趕緊把積壓的消息消費完，就得考慮提高消費能力，一般有兩種辦法：

• 消費者擴容：如果當前Topic的Message Queue的數量大於消費者數量，就可以對消費者進行擴容，增加消費者，來提高消費能力，儘快把積壓的消息消費玩。
• 消息遷移Queue擴容：如果當前Topic的Message Queue的數量小於或者等於消費者數量，這種情況，再擴容消費者就沒什麼用，就得考慮擴容Message Queue。可以新建一個臨時的Topic，臨時的Topic多設置一些Message Queue，然後先用一些消費者把消費的數據丟到臨時的Topic，因為不用業務處理，只是轉發一下消息，還是很快的。接下來用擴容的消費者去消費新的Topic里的數據，消費完了之後，恢復原狀。

3.2 如果消費者一直消費失敗導致消息積壓怎麼處理？

我們可以從以下幾個角度來考慮：

1、消費者出錯，肯定是程序或者其他問題導致的，如果容易修復，先把問題修復，讓consumer恢復正常消費。

2、如果時間來不及處理很麻煩，做轉發處理，寫一個臨時的consumer消費方案，先把消息消費，然後再轉發到一個新的topic和MQ資源，這個新的topic的機器資源單獨申請，要能承載住當前積壓的消息。

3、處理完積壓數據後，修復consumer，去消費新的MQ和現有的MQ數據，新MQ消費完成後恢復原狀。

 

MQ

3.3 那如果消息積壓達到磁盤上限，消息被刪除了怎麼辦？

最初，我們發送的消息記錄是落庫保存了的，而轉發發送的數據也保存了，那麼我們就可以通過這部分數據來找到丟失的那部分數據，再單獨跑個腳本重發就可以了。

如果轉發的程序沒有落庫，那就和消費方的記錄去做對比，只是過程會更艱難一點。

 4 順序消息如何實現？

順序消息是指消息的消費順序和產生順序相同，在有些業務邏輯下，必須保證順序，比如訂單的生成、付款、發貨，這個消息必須按順序處理才行。

 

順序消息分為全局順序消息和部分順序消息，全局順序消息指某個 Topic 下的所有消息都要保證順序；

部分順序消息只要保證每一組消息被順序消費即可，比如訂單消息，只要保證同一個訂單 ID 個消息能按順序消費即可。

部分順序消息

部分順序消息相對比較好實現，生產端需要做到把同 ID 的消息發送到同一個 Message Queue ；在消費過程中，要做到從同一個Message Queue讀取的消息順序處理——消費端不能並發處理順序消息，這樣才能達到部分有序。

 發送端使用 MessageQueueSelector 類來控制 把消息發往哪個 Message Queue 。

 消費端通過使用 MessageListenerOrderly 來解決單 Message Queue 的消息被並發處理的問題。

全局順序消息

RocketMQ 默認情況下不保證順序，比如創建一個 Topic ，默認八個寫隊列，八個讀隊列，這時候一條消息可能被寫入任意一個隊列里；在數據的讀取過程中，可能有多個 Consumer ，每個 Consumer 也可能啟動多個線程並行處理，所以消息被哪個 Consumer 消費，被消費的順序和寫人的順序是否一致是不確定的。

要保證全局順序消息， 需要先把 Topic 的讀寫隊列數設置為 一，然後Producer Consumer 的並發設置，也要是一。簡單來說，為了保證整個 Topic全局消息有序，只能消除所有的並發處理，各部分都設置成單線程處理 ，這時候就完全犧牲RocketMQ的高並發、高吞吐的特性了。