本文整理自滴滴出行消息隊列負責人 江海挺 在Apache RocketMQ開發者沙龍北京站的分享。
來源：公衆號阿里巴巴中間件

通過本文，您將瞭解到滴滴出行：

1. 在消息隊列技術選型方面的思考；2. 爲什麼選擇 RocketMQ 作爲出行業務的消息隊列解決方案；3. 如何構建自己的消息隊列服務；4. 在 RocketMQ 上的擴展改造實踐；5. 在 RocketMQ 上的實踐經驗。

江海挺：滴滴出行消息隊列負責人，Apache RocketMQ Contributor，大學畢業後一直在做消息隊列領域相關的技術、產品和服務，積累了豐富的實踐經驗，沉澱了不少關於消息隊列的思考。

滴滴出行的消息技術選型

1.1 歷史

初期，公司內部沒有專門的團隊維護消息隊列服務，所以消息隊列使用方式較多，主要以Kafka爲主，有業務直連的，也有通過獨立的服務轉發消息的。另外有一些團隊也會用RocketMQ、Redis的list，甚至會用比較非主流的beanstalkkd。導致的結果就是，比較混亂，無法維護，資源使用也很浪費。

1.2 爲什麼棄用 Kafka

一個核心業務在使用Kafka的時候，出現了集羣數據寫入抖動非常嚴重的情況，經常會有數據寫失敗。

主要有兩點原因：

• 隨着業務增長，Topic的數據增多，集羣負載增大，性能下降；
• 我們用的是Kafka0.8.2那個版本，有個bug，會導致副本重新複製，複製的時候有大量的讀，我們存儲盤用的是機械盤，導致磁盤IO過大，影響寫入。

所以我們決定做自己的消息隊列服務。

首先需要解決業務方消息生產失敗的問題。因爲這個Kafka用的是發佈/訂閱模式，一個topic的訂閱方會有很多，涉及到的下游業務也就非常多，沒辦法一口氣直接替換Kafka，遷移到新的一個消息隊列服務上。所以我們當時的方案是加了一層代理，然後利用codis作爲緩存，解決了Kafka不定期寫入失敗的問題，如上圖。當後面的Kafka出現不可寫入的時候，我們就會先把數據寫入到codis中，然後延時進行重試，直到寫成功爲止。

1.3 爲什麼選擇 RocketMQ

經過一系列的調研和測試之後，我們決定採用RocketMQ，具體原因在後面會介紹。

爲了支持多語言環境、解決一些遷移和某些業務的特殊需求，我們又在消費側加上了一個代理服務。然後形成了這麼一個核心框架。業務端只跟代理層交互。中間的消息引擎，負責消息的核心存儲。在之前的基本框架之後，我們後面就主要圍繞三個方向做。

• 遷移，把之前提到的所有五花八門的隊列環境，全部遷移到我們上面。這裏面的遷移方案後面會跟大家介紹一下。
• 功能迭代和成本性能上的優化。
• 服務化，業務直接通過平臺界面來申請資源，申請到之後直接使用。

1.4 演進中的架構

這張圖是我們消息隊列服務的一個比較新的現狀。先縱向看，上面是生產的客戶端，包括了7種語言。然後是我們的生產代理服務。在中間的是我們的消息存儲層。目前主要的消息存儲引擎是RocketMQ。然後還有一些在遷移過程中的Kafka。另一個是Chronos，它是我們延遲消息的一個存儲引擎。

再下面就是消費代理。消費代理同樣提供了多種語言的客戶端，還支持多種協議的消息主動推送功能，包括HTTP 協議 RESTful方式。結合我們的groovy腳本功能，還能實現將消息直接轉存到Redis、Hbase和HDFS上。此外，我們還在陸續接入更多的下游存儲。

除了存儲系統之外，我們也對接了實時計算平臺，例如Flink，Spark，Storm，左邊是我們的用戶控制檯和運維控制檯。這個是我們服務化的重點。用戶在需要使用隊列的時候，就通過界面申請Topic，填寫各種信息，包括身份信息，消息的峯值流量，消息大小，消息格式等等。然後消費方通過我們的界面，就可以申請消費。

運維控制檯，主要負責我們集羣的管理，自動化部署，流量調度，狀態顯示之類的功能。最後所有運維和用戶操作會影響線上的配置，都會通過ZooKeeper進行同步。

爲什麼選擇RocketMQ

我們圍繞以下兩個緯度進行了對比測試，結果顯示RocketMQ的效果更好。

2.1 測試-topic數量的支持

測試環境：Kafka 0.8.2，RocketMQ 3.4.6，1.0 Gbps Network，16 threads

測試結果如下：

這張圖是Kafka和RocketMQ在不同topic數量下的吞吐測試。橫座標是每秒消息數，縱座標是測試case。同時覆蓋了有無消費，和不同消息體的場景。一共8組測試數據，每組數據分別在Topic個數爲16、32、64、128、256時獲得的，每個topic包括8個Partition。下面四組數據是發送消息大小爲128字節的情況，上面四種是發送2k消息大小的情況。on 表示消息發送的時候，同時進行消息消費，off表示僅進行消息發送。

先看最上面一組數據，用的是Kafka，開啓消費，每條消息大小爲2048字節可以看到，隨着Topic數量增加，到256 Topic之後，吞吐極具下降。第二組是是RocketMQ。可以看到，Topic增大之後，影響非常小。第三組和第四組，是上面兩組關閉了消費的情況。結論基本類似，整體吞吐量會高那麼一點點。

下面的四組跟上面的區別是使用了128字節的小消息體。可以看到，Kafka吞吐受Topic數量的影響特別明顯。對比來看，雖然topic比較小的時候，RocketMQ吞吐較小，但是基本非常穩定，對於我們這種共享集羣來說比較友好。

2.2 測試-延遲

• Kafka

測試環境：Kafka 0.8.2.2，topic=1/8/32，Ack=1/all，replica=3

測試結果：

（橫座標對應吞吐，縱座標對應延遲時間）

上面的一組的3條線對應Ack=3，需要3個備份都確認後才完成數據的寫入。下面的一組的3條線對應Ack=1，有1個備份收到數據後就可以完成寫入。可以看到下面一組只需要主備份確認的寫入，延遲明顯較低。每組的三條線之間主要是Topic數量的區別，Topic數量增加，延遲也增大了。

• RocketMQ

測試環境：

RocketMQ 3.4.6，brokerRole=ASYNC/SYNC_MASTER, 2 Slave，

flushDiskType=SYNC_FLUSH/ASYNC_FLUSH

測試結果：

上面兩條是同步刷盤的情況，延遲相對比較高。下面的是異步刷盤。橙色的線是同步主從，藍色的線是異步主從。然後可以看到在副本同步複製的情況下，即橙色的線，4w的TPS之內都不超過1ms。用這條橙色的線和上面Kafka的圖中的上面三條線橫向比較來看，Kafka超過1w TPS 就超過1ms了。Kafka的延遲明顯更高。

如何構建自己的消息隊列

3.1 問題與挑戰

面臨的挑戰（順時針看）

• 客戶端語言，需要支持PHP、Go、Java、C++；
• 只有3個開發人員；
• 決定用RocketMQ，但是沒看過源碼；
• 上線時間緊，線上的Kafka還有問題；
• 可用性要求高。

使用RocketMQ時的兩個問題：

• 客戶端語言支持不全，以Java爲主，而我們還需要支持PHP、Go、C++；
• 功能特別多，如tag、property、消費過濾、RETRYtopic、死信隊列、延遲消費之類的功能，但這對我們穩定性維護來說，挑戰非常大。

針對以上兩個問題的解決辦法，如下圖所示：

• 使用ThriftRPC框架來解決跨語言的問題；
• 簡化調用接口。可以認爲只有兩個接口，send用來生產，pull用來消費。

主要策略就是堅持KISS原則（Keep it simple, stupid），保持簡單，先解決最主要的問題，讓消息能夠流轉起來。然後我們把其他主要邏輯都放在了proxy這一層來做，比如限流、權限認證、消息過濾、格式轉化之類的。這樣，我們就能儘可能地簡化客戶端的實現邏輯，不需要把很多功能用各種語言都寫一遍。

3.2 遷移方案

架構確定後，接下來是我們的一個遷移過程。

遷移這個事情，在pub-sub的消息模型下，會比較複雜。因爲下游的數據消費方可能很多，上游的數據沒法做到一刀切流量，這就會導致整個遷移的週期特別長。然後我們爲了儘可能地減少業務遷移的負擔，加快遷移的效率，我們在Proxy層提供了雙寫和雙讀的功能。

• 雙寫：ProcucerProxy同時寫RocketMQ和Kafka；
• 雙讀：ConsumerProxy同時從RocketMQ和Kafka消費數據。

有了這兩個功能之後，我們就能提供以下兩種遷移方案了。

3.2.1 雙寫

生產端雙寫，同時往Kafka和RocketMQ寫同樣的數據，保證兩邊在整個遷移過程中都有同樣的全量數據。Kafka和RocketMQ有相同的數據，這樣下游的業務也就可以開始遷移。如果消費端不關心丟數據，那麼可以直接切換，切完直接更新消費進度。如果需要保證消費必達，可以先在ConsumerProxy設置消費進度，消費客戶端保證沒有數據堆積後再去遷移，這樣會有一些重複消息，一般客戶端會保證消費處理的冪等。

生產端的雙寫其實也有兩種方案：

• 客戶端雙寫，如下圖：

業務那邊不停原來的kafka 客戶端。只是加上我們的客戶端，往RocketMQ裏追加寫。這種方案在整個遷移完成之後，業務還需要把老的寫入停掉。相當於兩次上線。

• Producer Proxy雙寫，如下圖：

業務方直接切換生產的客戶端，只往我們的proxy上寫數據。然後我們的proxy負責把數據複製，同時寫到兩個存儲引擎中。這樣在遷移完成之後，我們只需要在Proxy上關掉雙寫功能就可以了。對生產的業務方來說是無感知的，生產方全程只需要改造一次，上一下線就可以了。

所以表面看起來，應該還是第二種方案更加簡單。但是，從整體可靠性的角度來看，一般還是認爲第一種相對高一點。因爲客戶端到Kafka這一條鏈路，業務之前都已經跑穩定了。一般不會出問題。但是寫我們Proxy就不一定了，在接入過程中，是有可能出現一些使用上的問題，導致數據寫入失敗，這就對業務方測試質量的要求會高一點。然後消費的遷移過程，其實風險是相對比較低的。出問題的時候，可以立即回滾。因爲它在老的Kafka上消費進度，是一直保留的，而且在遷移過程中，可以認爲是全量雙消費。

以上就是數據雙寫的遷移方案，這種方案的特點就是兩個存儲引擎都有相同的全量數據。

3.2.2 雙讀

特點：保證不會重複消費。對於P2P 或者消費下游不太多，或者對重複消費數據比較敏感的場景比較適用。

這個方案的過程是這樣的，消費先切換。全部遷移到到我們的Proxy上消費，Proxy從Kafka上獲取。這個時候RocketMQ上沒有流量。但是我們的消費Proxy保證了雙消費，一旦RocketMQ有流量了，客戶端同樣也能收到。然後生產方改造客戶端，直接切流到RocketMQ中，這樣就完成了整個流量遷移過程。運行一段時間，比如Kafka裏的數據都過期之後，就可以把消費Proxy上的雙消費關了，下掉Kafka集羣。

整個過程中，生產直接切流，所以數據不會重複存儲。然後在消費遷移的過程中，我們消費Proxy上的group和業務原有的group可以用一個名字，這樣就能實現遷移過程中自動rebalance，這樣就能實現沒有大量重複數據的效果。所以這個方案對重複消費比較敏感的業務會比較適合的。這個方案的整個過程中，消費方和生產方都只需要改造一遍客戶端，上一次線就可以完成。

RocketMQ擴展改造

說完遷移方案，這裏再簡單介紹一下，我們在自己的RocketMQ分支上做的一些比較重要的事情。

首先一個非常重要的一點是主從的自動切換。

熟悉RocketMQ的同學應該知道，目前開源版本的RocketMQ broker 是沒有主從自動切換的。如果你的Master掛了，那你就寫不進去了。然後slave只能提供只讀的功能。當然如果你的topic在多個主節點上都創建了，雖然不會完全寫不進去，但是對單分片順序消費的場景，還是會產生影響。所以呢，我們就自己加了一套主從自動切換的功能。

第二個是批量生產的功能。

RocketMQ4.0之後的版本是支持批量生產功能的。但是限制了，只能是同一個ConsumerQueue的。這個對於我們的Proxy服務來說，不太友好，因爲我們的proxy是有多個不同的topic的，所以我們就擴展了一下，讓它能夠支持不同Topic、不同Consume Queue。原理上其實差不多，只是在傳輸的時候，把Topic和Consumer Queue的信息都編碼進去。

第三個，元信息管理的改造。

目前RocketMQ單機能夠支持的Topic數量，基本在幾萬這麼一個量級，在增加上去之後，元信息的管理就會非常耗時，對整個吞吐的性能影響相對來說就會非常大。然後我們有個場景又需要支持單機百萬左右的Topic數量，所以我們就改造了一下元信息管理部分，讓RocketMQ單機能夠支撐的Topic數量達到了百萬。

後面一些就不太重要了，比如集成了我們公司內部的一些監控和部署工具，修了幾個bug，也給提了PR。最新版都已經fix掉了。

RocketMQ使用經驗

接下來，再簡單介紹一下，我們在RocketMQ在使用和運維上的一些經驗。主要是涉及在磁盤IO性能不夠的時候，一些參數的調整。

5.1 讀老數據的問題

我們都知道，RocketMQ的數據是要落盤的，一般只有最新寫入的數據纔會在PageCache中。比如下游消費數據，因爲一些原因停了一天之後，又突然起來消費數據。這個時候就需要讀磁盤上的數據。然後RocketMQ的消息體是全部存儲在一個append only的 commitlog 中的。如果這個集羣中混雜了很多不同topic的數據的話，要讀的兩條消息就很有可能間隔很遠。最壞情況就是一次磁盤IO讀一條消息。這就基本等價於隨機讀取了。如果磁盤的IOPS（Input/Output Operations Per Second）扛不住，還會影響數據的寫入，這個問題就嚴重了。

值得慶幸的是，RocketMQ提供了自動從Slave讀取老數據的功能。這個功能主要由slaveReadEnable這個參數控制。默認是關的（slaveReadEnable = false bydefault）。推薦把它打開，主從都要開。這個參數打開之後，在客戶端消費數據時，會判斷，當前讀取消息的物理偏移量跟最新的位置的差值，是不是超過了內存容量的一個百分比（accessMessageInMemoryMaxRatio= 40 by default）。如果超過了，就會告訴客戶端去備機上消費數據。如果採用異步主從，也就是brokerRole等於ASYNC_AMSTER的時候，你的備機IO打爆，其實影響不太大。但是如果你採用同步主從，那還是有影響。所以這個時候，最好掛兩個備機。因爲RocketMQ的主從同步複製，只要一個備機響應了確認寫入就可以了，一臺IO打爆，問題不大。

5.2 過期數據刪除

RocketMQ默認數據保留72個小時（fileReservedTime=72）。然後它默認在凌晨4點開始刪過期數據（deleteWhen="04"）。你可以設置多個值用分號隔開。因爲數據都是定時刪除的，所以在磁盤充足的情況，數據的最長保留會比你設置的還多一天。又由於默認都是同一時間，刪除一整天的數據，如果用了機械硬盤，一般磁盤容量會比較大，需要刪除的數據會特別多，這個就會導致在刪除數據的時候，磁盤IO被打滿。這個時候又要影響寫入了。

爲了解決這個問題，可以嘗試多個方法，一個是設置文件刪除的間隔，有兩個參數可以設置，

• deleteCommitLogFilesInterval = 100（毫秒）。每刪除10個commitLog文件的時間間隔；
• deleteConsumeQueueFilesInterval=100（毫秒）。每刪除一個ConsumeQueue文件的時間間隔。

另外一個就是增加刪除頻率，把00-23都寫到deleteWhen，就可以實現每個小時都刪數據。

5.3 索引

默認情況下，所有的broker都會建立索引（messageIndexEnable=true）。這個索引功能可以支持按照消息的uniqId，消息的key來查詢消息體。索引文件實現的時候，本質上也就是基於磁盤的個一個hashmap。如果broker上消息數量比較多，查詢的頻率比較高，這也會造成一定的IO負載。所以我們的推薦方案是在Master上關掉了index功能，只在slave上打開。然後所有的index查詢全部在slave上進行。當然這個需要簡單修改一下MQAdminImpl裏的實現。因爲默認情況下，它會向Master發出請求。