你應該知道的RocketMQ

時間 2021-08-14

標籤 mysql 算法 sql markdown 網絡架構併發框架異步 socket 欄目 MySQL 简体版

原文原文鏈接

1.概述

在好久以前寫過一篇Kafka相關的文章，你須要知道的Kafka，那個時候在業務上更多的是使用的是Kafka，而如今換了公司以後，更多的使用的是Rocketmq，本篇文章會盡力全面的介紹RocketMQ和Kafka各個關鍵點的比較，但願你們讀完能有所收穫。mysql

RocketMQ前身叫作MetaQ, 在MeataQ發佈3.0版本的時候更名爲RocketMQ，其本質上的設計思路和Kafka相似，可是和Kafka不一樣的是其使用Java進行開發，因爲在國內的Java受衆羣體遠遠多於Scala，因此RocketMQ是不少以Java語言爲主的公司的首選。一樣的RocketMQ和Kafka都是Apache基金會中的頂級項目，他們社區的活躍度都很是高，項目更新迭代也很是快。算法

2.入門實例

2.1 生產者

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {

        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();

        for (int i = 0; i < 128; i++)
            try {
                {
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }

        producer.shutdown();
    }
}

直接定義好一個producer，建立好Message，調用send方法便可。sql

2.2 消費者

public class PushConsumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
        consumer.subscribe("TopicTest", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //wrong time format 2017_0422_221800
        consumer.setConsumeTimestamp("20181109221800");
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

3.RocketMQ架構原理

對於RocketMQ先拋出幾個問題:markdown

RocketMQ的topic和隊列是什麼樣的，和Kafka的分區有什麼不一樣？
RocketMQ網絡模型是什麼樣的，和Kafka對好比何？
RocketMQ消息存儲模型是什麼樣的，如何保證高可靠的存儲，和Kafka對好比何？

3.1 RocketMQ架構圖

對於RocketMQ的架構圖，在大致上來看和Kafka並無太多的差異，可是在不少細節上是有不少差異的，接下來會一一進行講述。網絡

3.2 RocketMQ名詞解釋

在3.1的架構中咱們有多個Producer，多個主Broker，多個從Broker,每一個Producer能夠對應多個Topic,每一個Consumer也能夠消費多個Topic。架構

Broker信息會上報至NameServer,Consumer會從NameServer中拉取Broker和Topic的信息。併發

Producer：消息生產者，向Broker發送消息的客戶端
Consumer：消息消費者，從Broker讀取消息的客戶端
Broker：消息中間的處理節點，這裏和kafka不一樣，kafka的Broker沒有主從的概念，均可以寫入請求以及備份其餘節點數據，RocketMQ只有主Broker節點才能寫，通常也經過主節點讀，當主節點有故障或者一些其餘特殊狀況纔會使用從節點讀，有點相似- 於mysql的主從架構。
Topic：消息主題，一級消息類型，生產者向其發送消息, 消費者讀取其消息。
Group：分爲ProducerGroup,ConsumerGroup,表明某一類的生產者和消費者，通常來講同一個服務能夠做爲Group,同一個Group通常來講發送和消費的消息都是同樣的。
Tag：Kafka中沒有這個概念，Tag是屬於二級消息類型，通常來講業務有關聯的可使用同一個Tag,好比訂單消息隊列，使用Topic_Order,Tag能夠分爲Tag_食品訂單,Tag_服裝訂單等等。
Queue: 在kafka中叫Partition,每一個Queue內部是有序的，在RocketMQ中分爲讀和寫兩種隊列，通常來講讀寫隊列數量一致，若是不一致就會出現不少問題。
NameServer：Kafka中使用的是ZooKeeper保存Broker的地址信息，以及Broker的Leader的選舉，在RocketMQ中並無採用選舉Broker的策略，因此採用了無狀態的NameServer來存儲，因爲NameServer是無狀態的，集羣節點之間並不會通訊，因此上傳數據的時候都須要向全部節點進行發送。

不少朋友都在問什麼是無狀態呢？狀態的有無實際上就是數據是否會作存儲，有狀態的話數據會被持久化，無狀態的服務能夠理解就是一個內存服務，NameServer自己也是一個內存服務，全部數據都存儲在內存中，重啓以後都會丟失。框架

3.3 Topic和Queue

在RocketMQ中的每一條消息，都有一個Topic，用來區分不一樣的消息。一個主題通常會有多個消息的訂閱者，當生產者發佈消息到某個主題時，訂閱了這個主題的消費者均可以接收到生產者寫入的新消息。異步

在Topic中有分爲了多個Queue，這實際上是咱們發送/讀取消息通道的最小單位，咱們發送消息都須要指定某個寫入某個Queue，拉取消息的時候也須要指定拉取某個Queue，因此咱們的順序消息能夠基於咱們的Queue維度保持隊列有序，若是想作到全局有序那麼須要將Queue大小設置爲1，這樣全部的數據都會在Queue中有序。
socket

在上圖中咱們的Producer會經過一些策略進行Queue的選擇：

非順序消息：非順序消息通常直接採用輪訓發送的方式進行發送。
順序消息：根據某個Key好比咱們常見的訂單Id,用戶Id，進行Hash，將同一類數據放在同一個隊列中，保證咱們的順序性。

咱們同一組Consumer也會根據一些策略來選Queue，常見的好比平均分配或者一致性Hash分配。

要注意的是當Consumer出現下線或者上線的時候，這裏須要作重平衡，也就是Rebalance，RocketMQ的重平衡機制以下:

定時拉取broker,topic的最新信息
每隔20s作重平衡
隨機選取當前Topic的一個主Broker，這裏要注意的是否是每次重平衡全部主Broker都會被選中，由於會存在一個Broker再多個Broker的狀況。
獲取當前Broker，當前ConsumerGroup的全部機器ID。
而後進行策略分配。

因爲重平衡是定時作的，因此這裏有可能會出現某個Queue同時被兩個Consumer消費，因此會出現消息重複投遞。

Kafka的重平衡機制和RocketMQ不一樣，Kafka的重平衡是經過Consumer和Coordinator聯繫來完成的，當Coordinator感知到消費組的變化，會在心跳過程當中發送重平衡的信號，而後由一個ConsumerLeader進行重平衡選擇，而後再由Coordinator將結果通知給全部的消費者。

3.3.1 Queue讀寫數量不一致

在RocketMQ中Queue被分爲讀和寫兩種，在最開始接觸RocketMQ的時候一直覺得讀寫隊列數量配置不一致不會出現什麼問題的，好比當消費者機器不少的時候咱們配置不少讀的隊列，可是實際過程當中發現會出現消息沒法消費和根本沒有消息消費的狀況。

當寫的隊列數量大於讀的隊列的數量，當大於讀隊列這部分ID的寫隊列的數據會沒法消費，由於不會將其分配給消費者。
當讀的隊列數量大於寫的隊列數量，那麼多的隊列數量就不會有消息被投遞進來。

這個功能在RocketMQ在我看來明顯沒什麼用，由於基本上都會設置爲讀寫隊列大小同樣，這個爲啥不直接將其進行統一，反而容易讓用戶配置不同出現錯誤。

這個問題在RocketMQ的Issue裏也沒有收到好的答案。

3.4 消費模型

通常來講消息隊列的消費模型分爲兩種，基於推送的消息(push)模型和基於拉取(poll)的消息模型。

基於推送模型的消息系統，由消息代理記錄消費狀態。消息代理將消息推送到消費者後，標記這條消息爲已經被消費，可是這種方式沒法很好地保證消費的處理語義。好比當咱們把已經把消息發送給消費者以後，因爲消費進程掛掉或者因爲網絡緣由沒有收到這條消息，若是咱們在消費代理將其標記爲已消費，這個消息就永久丟失了。若是咱們利用生產者收到消息後回覆這種方法，消息代理須要記錄消費狀態，這種不可取。

用過RocketMQ的同窗確定不由會想到，在RocketMQ中不是提供了兩種消費者嗎？
MQPullConsumer和MQPushConsumer，其中MQPushConsumer不就是咱們的推模型嗎？其實這兩種模型都是客戶端主動去拉消息，其中的實現區別以下：

MQPullConsumer：每次拉取消息須要傳入拉取消息的offset和每次拉取多少消息量，具體拉取哪裏的消息，拉取多少是由客戶端控制。
MQPushConsumer：一樣也是客戶端主動拉取消息，可是消息進度是由服務端保存，Consumer會定時上報本身消費到哪裏，因此Consumer下次消費的時候是能夠找到上次消費的點，通常來講使用PushConsumer咱們不須要關心offset和拉取多少數據，直接使用便可。

3.4.1 集羣消費和廣播消費

消費模式咱們分爲兩種，集羣消費，廣播消費:

集羣消費: 同一個GroupId都屬於一個集羣，通常來講一條消息只會被任意一個消費者處理。
廣播消費：廣播消費的消息會被集羣中全部消費者進行消息，可是要注意一下由於廣播消費的offset在服務端保存成本過高，因此客戶端每一次重啓都會從最新消息消費，而不是上次保存的offset。

3.5 網絡模型

在Kafka中使用的原生的socket實現網絡通訊，而RocketMQ使用的是Netty網絡框架，如今愈來愈多的中間件都不會直接選擇原生的socket，而是使用的Netty框架，主要得益於下面幾個緣由:

API使用簡單，不須要關心過多的網絡細節，更專一於中間件邏輯。
性能高。
成熟穩定，jdk nio的bug都被修復了。

選擇框架是一方面，而想要保證網絡通訊的高效，網絡線程模型也是一方面，咱們常見的有1+N(1個Acceptor線程，N個IO線程)，1+N+M(1個acceptor線程，N個IO線程，M個worker線程)等模型，RocketMQ使用的是1+N1+N2+M的模型，以下圖所示：

1個acceptor線程，N1個IO線程，N2個線程用來作Shake-hand,SSL驗證,編解碼;M個線程用來作業務處理。這樣的好處將編解碼，和SSL驗證等一些可能耗時的操做放在了一個單獨的線程池，不會佔據咱們業務線程和IO線程。

3.6 高可靠的分佈式存儲模型

作爲一個好的消息系統，高性能的存儲，高可用都不可少。

3.6.1 高性能日誌存儲

RocketMQ和Kafka的存儲核心設計有很大的不一樣，因此其在寫入性能方面也有很大的差異，這是16年阿里中間件團隊對RocketMQ和Kafka不一樣Topic下作的性能測試:

從圖上能夠看出：

Kafka在Topic數量由64增加到256時，吞吐量降低了98.37%。
RocketMQ在Topic數量由64增加到256時，吞吐量只降低了16%。
這是爲何呢？kafka一個topic下面的全部消息都是以partition的方式分佈式的存儲在多個節點上。同時在kafka的機器上，每一個Partition其實都會對應一個日誌目錄，在目錄下面會對應多個日誌分段。因此若是Topic不少的時候Kafka雖然寫文件是順序寫，但實際上文件過多，會形成磁盤IO競爭很是激烈。

那RocketMQ爲何在多Topic的狀況下，依然還能很好的保持較多的吞吐量呢？咱們首先來看一下RocketMQ中比較關鍵的文件:

這裏有四個目錄（這裏的解釋就直接用RocketMQ官方的了）:

commitLog：消息主體以及元數據的存儲主體，存儲Producer端寫入的消息主體內容,消息內容不是定長的。單個文件大小默認1G ，文件名長度爲20位，左邊補零，剩餘爲起始偏移量，好比00000000000000000000表明了第一個文件，起始偏移量爲0，文件大小爲1G=1073741824；當第一個文件寫滿了，第二個文件爲00000000001073741824，起始偏移量爲1073741824，以此類推。消息主要是順序寫入日誌文件，當文件滿了，寫入下一個文件；
config：保存一些配置信息，包括一些Group，Topic以及Consumer消費offset等信息。
consumeQueue:消息消費隊列，引入的目的主要是提升消息消費的性能，因爲RocketMQ是基於主題topic的訂閱模式，消息消費是針對主題進行的，若是要遍歷commitlog文件中根據topic檢索消息是很是低效的。Consumer便可根據ConsumeQueue來查找待消費的消息。其中，ConsumeQueue（邏輯消費隊列）做爲消費消息的索引，保存了指定Topic下的隊列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件能夠當作是基於topic的commitlog索引文件，故consumequeue文件夾的組織方式以下：topic/queue/file三層組織結構，具體存儲路徑爲：

HOME \store\index\${fileName}，文件名fileName是以建立時的時間戳命名的，固定的單個IndexFile文件大小約爲400M，一個IndexFile能夠保存 2000W個索引，IndexFile的底層存儲設計爲在文件系統中實現HashMap結構，故rocketmq的索引文件其底層實現爲hash索引。

咱們發現咱們的消息主體數據並無像Kafka同樣寫入多個文件，而是寫入一個文件,這樣咱們的寫入IO競爭就很是小，能夠在不少Topic的時候依然保持很高的吞吐量。有同窗說這裏的ConsumeQueue寫是在不停的寫入呢，而且ConsumeQueue是以Queue維度來建立文件，那麼文件數量依然不少，在這裏ConsumeQueue的寫入的數據量很小，每條消息只有20個字節，30W條數據也才6M左右，因此其實對咱們的影響相對Kafka的Topic之間影響是要小不少的。咱們整個的邏輯能夠以下：

Producer不斷的再往CommitLog添加新的消息，有一個定時任務ReputService會不斷的掃描新添加進來的CommitLog，而後不斷的去構建ConsumerQueue和Index。

注意：這裏指的都是普通的硬盤，在SSD上面多個文件併發寫入和單個文件寫入影響不大。

讀取消息

Kafka中每一個Partition都會是一個單獨的文件，因此當消費某個消息的時候，會很好的出現順序讀，咱們知道OS從物理磁盤上訪問讀取文件的同時，會順序對其餘相鄰塊的數據文件進行預讀取，將數據放入PageCache，因此Kafka的讀取消息性能比較好。

RocketMQ讀取流程以下：

先讀取ConsumerQueue中的offset對應CommitLog物理的offset
根據offset讀取CommitLog

ConsumerQueue也是每一個Queue一個單獨的文件，而且其文件體積小，因此很容易利用PageCache提升性能。而CommitLog，因爲同一個Queue的連續消息在CommitLog實際上是不連續的，因此會形成隨機讀，RocketMQ對此作了幾個優化：

Mmap映射讀取，Mmap的方式減小了傳統IO將磁盤文件數據在操做系統內核地址空間的緩衝區和用戶應用程序地址空間的緩衝區之間來回進行拷貝的性能開銷
使用DeadLine調度算法+SSD存儲盤
因爲Mmap映射受到內存限制，當不在Mmmap映射這部分數據的時候(也就是消息堆積過多)，默認是內存的40%，會將請求發送到SLAVE,減緩Master的壓力
3.6.2 可用性

3.6.2.1 集羣模式

咱們首先須要選擇一種集羣模式，來適應咱們可忍耐的可用程度，通常來講分爲三種：

單Master:這種模式，可用性最低，可是成本也是最低，一旦宕機，全部都不可用。這種通常只適用於本地測試。
單Master多SLAVE:這種模式，可用性通常，若是主宕機，那麼全部寫入都不可用，讀取依然可用，若是master磁盤損壞，能夠依賴slave的數據。
多Master:這種模式，可用性通常，若是出現部分master宕機，那麼這部分master上的消息都不可消費，也不可寫數據，若是一個Topic的隊列在多個Master上都有，那麼能夠保證沒有宕機的那部分能夠正常消費，寫入。若是master的磁盤損壞會致使消息丟失。
多Master多Slave：這種模式，可用性最高，可是維護成本也最高，當master宕機了以後，只會出如今這部分master上的隊列不可寫入，可是讀取依然是能夠的，而且若是master磁盤損壞，能夠依賴slave的數據。

通常來講投入生產環境的話都會選擇第四種，來保證最高的可用性。

3.6.2.2 消息的可用性

當咱們選擇好了集羣模式以後，那麼咱們須要關心的就是怎麼去存儲和複製這個數據，rocketMQ對消息的刷盤提供了同步和異步的策略來知足咱們的，當咱們選擇同步刷盤以後，若是刷盤超時會給返回FLUSH_DISK_TIMEOUT，若是是異步刷盤不會返回刷盤相關信息，選擇同步刷盤能夠盡最大程度知足咱們的消息不會丟失。

除了存儲有選擇以後，咱們的主從同步提供了同步和異步兩種模式來進行復制，固然選擇同步能夠提高可用性，可是消息的發送RT時間會降低10%左右。

3.6.3 Dleger

咱們上面對於master-slave部署模式已經作了不少分析，咱們發現，當master出現問題的時候，咱們的寫入怎麼都會不可用，除非恢復master，或者手動將咱們的slave切換成master，致使了咱們的Slave在多數狀況下只有讀取的做用。RocketMQ在最近的幾個版本中推出了Dleger-RocketMQ，使用Raft協議複製CommitLog，而且自動進行選主，這樣master宕機的時候，寫入依然保持可用。

有關Dleger-RocketMQ的信息更多的能夠查看這篇文章：Dledger-RocketMQ 基於Raft協議的commitlog存儲庫。

3.7 定時/延時消息

定時消息和延時消息在實際業務場景中使用的比較多，好比下面的一些場景：

訂單超時未支付自動關閉，由於在不少場景中下單以後庫存就被鎖定了，這裏須要將其進行超時關閉。
須要一些延時的操做，好比一些兜底的邏輯，當作完某個邏輯以後，能夠發送延時消息好比延時半個小時，進行兜底檢查補償。
在某個時間給用戶發送消息，一樣也可使用延時消息。

在開源版本的RocketMQ中延時消息並不支持任意時間的延時，須要設置幾個固定的延時等級，目前默認設置爲：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，從1s到2h分別對應着等級1到18，而阿里雲中的版本(要付錢)是能夠支持40天內的任什麼時候刻（毫秒級別）。咱們先看下在RocketMQ中定時任務原理圖：

Step1：Producer在本身發送的消息上設置好須要延時的級別。
Step2: Broker發現此消息是延時消息，將Topic進行替換成延時Topic，每一個延時級別都會做爲一個單獨的queue，將本身的Topic做爲額外信息存儲。
Step3: 構建ConsumerQueue
Step4: 定時任務定時掃描每一個延時級別的ConsumerQueue。
Step5: 拿到ConsumerQueue中的CommitLog的Offset，獲取消息，判斷是否已經達到執行時間
Step6: 若是達到，那麼將消息的Topic恢復，進行從新投遞。若是沒有達到則延遲沒有達到的這段時間執行任務。

能夠看見延時消息是利用新建單獨的Topic和Queue來實現的，若是咱們要實現40天以內的任意時間度，基於這種方案，那麼須要402460601000個queue，這樣的成本是很是之高的，那阿里雲上面的支持任意時間是怎麼實現的呢？這裏猜想是持久化二級TimeWheel時間輪，二級時間輪用於替代咱們的ConsumeQueue，保存Commitlog-Offset，而後經過時間輪不斷的取出當前已經到了的時間，而後再次投遞消息。具體的實現邏輯須要後續會單獨寫一篇文章。

3.8 事務消息

事務消息一樣的也是RocketMQ中的一大特點，其能夠幫助咱們完成分佈式事務的最終一致性，有關分佈式事務相關的能夠看我之前的不少文章都有不少詳細的介紹，這裏直接關注公衆號：咖啡拿鐵。

具體使用事務消息步驟以下：

Step1：調用sendMessageInTransaction發送事務消息
Step2: 若是發送成功，則執行本地事務。
Step3: 若是執行本地事務成功則發送commit，若是失敗則發送rollback。
Step4: 若是其中某個階段好比commit發送失敗，rocketMQ會進行定時從Broker回查，本地事務的狀態。

事務消息的使用整個流程相對以前幾種消息使用比較複雜，下面是事務消息實現的原理圖：

Step1: 發送事務消息,這裏也叫作halfMessage，會將Topic替換爲HalfMessage的Topic。
Step2: 發送commit或者rollback，若是是commit這裏會查詢出以前的消息，而後將消息復原成原Topic，而且發送一個OpMessage用於記錄當前消息能夠刪除。若是是rollback這裏會直接發送一個OpMessage刪除。
Step3: 在Broker有個處理事務消息的定時任務，定時對比halfMessage和OpMessage,若是有OpMessage且狀態爲刪除，那麼該條消息一定commit或者rollback，因此就能夠刪除這條消息。
Step4: 若是事務超時（默認是6s），尚未opMessage，那麼頗有可能commit信息丟了，這裏會去反查咱們的Producer本地事務狀態。
Step5: 根據查詢出來的信息作Step2。

咱們發現RocketMQ實現事務消息也是經過修改原Topic信息，和延遲消息同樣，而後模擬成消費者進行消費，作一些特殊的業務邏輯。固然咱們還能夠利用這種方式去作RocketMQ更多的擴展。