業務背景

今日頭條的服務大量使用微服務，容器數目巨大，業務線繁多， Topic 的數量也非常多。另外，使用的語言比較繁雜，包括 Python，Go, C++, Java, JS 等，對於基礎元件的接入，維護 SDK 的成本很高。

引入 RocketMQ 之前採用的訊息佇列是 NSQ 和 kafka ， NSQ 是純記憶體的訊息佇列，缺少訊息的永續性，不落盤直接寫到 Golang 的 channel 裡，在併發量高的時候 CPU 利用率非常高，其優點是可以無限水平擴充套件，另外，由於不需要保證訊息的有序性，叢集單點故障對可用性基本沒有影響，所以具有非常高的可用性。我們也用到了 Kafka ，它的主要問題是在業務線和 Topic 繁多，其寫入效能會出現明顯的下降，拆分叢集又會增加額外的運維負擔。並且在高負載下，其故障恢復時間比較長。所以，針對當時的狀況和業務場景的需求，我們進行了一些調研，期望選擇一款新的 MQ 來比較好的解決目前的困境，最終選擇了 RocketMQ

為什麼選擇 RocketMQ

這是一個經過阿里巴巴多年雙11驗證過的、可以支援億級併發的開源訊息佇列，是值得信任的。其次關注一下他的特性。 RocketMQ 具有高可靠性、資料永續性，和 Kafka 一樣是先寫 PageCache ，再落盤，並且資料有多副本；並且它的儲存模型是所有的 Topic 都寫到同一個 Commitlog 裡，是一個append only 操作，在海量 Topic 下也能將磁碟的效能發揮到極致，並且保持穩定的寫入時延。然後就是他的效能，經過我們的 benchmark ，採用一主兩從的結構，單機 qps 可以達到 14w , latency 保持在 2ms 以內。對比之前的 NSQ 和 Kafka ， Kafka 的吞吐非常高，但是在多 Topic 下， Kafka 的 PCT99 毛刺會非常多,而且平均值非常長，不適合線上業務場景。另外 NSQ 的訊息首先經過 Golang 的 channel ，這是非常消耗 CPU 的，在單機 5~6w 的時候 CPU 利用率達到 50~60% ，高負載下的寫延遲不穩定。另外 RocketMQ 對線上業務特性支援是非常豐富的，支援 retry , 支援併發消費，死信佇列，延時訊息，基於時間戳的訊息回溯，另外訊息體支援訊息頭，這個是非常有用的，可以直接支援實現訊息鏈路追蹤，不然就需要把追蹤資訊寫到 message 的 body 裡；還支援事務的訊息。綜合以上特性最終選擇了 RocketMQ 。

RocketMQ 在頭條的落地實踐

下面簡單介紹下，今日頭條的部署結構，如圖所示：

由於生產者種類繁多，我們傾向於保持客戶端簡單，因為推動 SDK 升級是一個很沉重的負擔，所以我們通過提供一個 Proxy 層，來保持生產端的輕量。 Proxy 層是由一個標準的 gRpc 框架實現，也可以用 thrift ,當然任何 RPC 都框架都可以實現。

Producer 的 Proxy 相對比較簡單，雖然在 Producer 這邊也集成了很多比如路由管理、監控等其他功能， SDK 只需實現發訊息的請求，所以 SDK 的非常輕量、改動非常少，在迭代過程中也不需要一個個推業務去升級 SDK 。 SDK 通過服務發現去找到一個 Proxy 例項，然後建立連線傳送訊息， Proxy 的工作是根據 RPC 請求的訊息轉發到對應的 Broker 叢集上。 Consumer Proxy 實現的是 pull 和二次 reblance 的邏輯，這個後面會講到，相當於把 Consumer 的 pull 透傳給 Brokerset ， Proxy 這邊會有一個訊息的 cache ，一定程度上降低對 broker page cache 的汙染。這個架構和滴滴的 MQ 架構有點相似，他們也是之前做了一個 Proxy ，用 thrift 做 RPC ，這對後端的擴容、運維、減少 SDK 的邏輯上來說都是很有必要的。

在容器以及微服務場景下為什麼要做這個 Porxy ?

有以下幾點原因：
1、 SDK 會非常簡單輕量。

2、很容易對流量進行控制； Proxy 可以對生產端的流量進行控制，比如我們期望某些Broker壓力比較大的時候，能夠切一些流量或者說切流量到另外的機房，這種流量的排程，多環境的支援，再比如有些預釋出環境、預上線環境的支援，我們 Topic 這邊寫入的流量可以在 Proxy 這邊可以很方便的完成控制，不用修改 SDK 。

3，解決連線的問題；特別是解決 Python 的問題， Python 實現的服務如果要獲得高併發度，一般是採取多程序模型，這意味著一個程序一個連線，特別是對於部署到 Docker 裡的 Python 服務，它可能一個容器裡啟動幾百個程序，如果直接連到 Broker ，這個 Broker 上的連線數可能到幾十上百萬，此時 CPU 軟中斷會非常高，導致讀寫的延時的明顯上漲

4，通過 Proxy ,多了一個代理，在消費不需要順序的情況下，我們可以支援更高的併發度， Consumer 的例項數可以超過 Consume Queue 的數量。

5，可以無縫的繼承其他的 MQ 。中間有一層 Proxy ,後面可以更改儲存引擎，這個對客戶端是無感知的。

6，在 Conusmer 在升級或 Restart 的時候， Consumer 如果直接連 broker 的話， rebalance 觸發比較頻繁， 如果 rebalance 比較頻繁，且 Topic 量比較大的時候，可能會造成訊息堆積，這個業務不是太接受的；如果加一層 Proxy 的話， rebalance 只在 Proxt 和 Broker 之間進行，就不需要 Consumer 再進行一次 rebalance ， Proxy 只需要維護著和自己建立連線的 Consumer 就可以了。當消費者重啟或升級的時候，可以最小程度的減少 rebalance 。

以上是我們通過 Proxy 介面給 RocketMQ 帶來的好處。因為多了一層，也會帶來額外的 Overhead 的，如下：
1，會消耗 CPU ， Proxy 那一層會做RPC協議的序列化和反序列化。
如下是 Conusme Proxy 的結構圖，它帶來了消費併發度的提高。由於我們的 Broker 叢集是獨立部署的，考慮到broker主要是消耗包括網絡卡、磁碟和記憶體資源，對於 CPU 的消耗反而不高，這裡的解決方式直接進行混合部署，然後直接在新的機器上進行擴，但是 Broker 這邊的 CPU 也是可以得到利用的。

2，延遲問題。經過測試，在 4Kmsg、20W Tps 下，延遲會有所增加，大概是 1ms ,從 2ms 到 3ms 左右，這個時延對於業務來說是可以接受的。

下面看下 Consumer 這邊的邏輯，如下圖所示，

比如上面部署了兩個 Proxy , Broker，左邊有 6 個 Queue ，對於順序訊息來說，左邊這邊 rebalance 是一個相對靜態的結果， Consumer 的上下線是比較頻繁的。對於順序訊息來說，左邊和之前的邏輯是保持一致的， Proxy 會為每個 Consumer 例項分配到合適的數量的 Queue ；對於不關心順序性的訊息，Proxy 會把所有的訊息都放到一個佇列裡，然後從這個佇列 dispatch 到各個 Consumer ，對於亂序訊息來說，理論上來說 Consumer 數量可以無限擴充套件的；相對於和普通 Consumer 直連的情況，Consumer 的數量如果超過了Consume Queue的數量，其中多出來的 Consumer 是沒有辦法分配到 Queue 的，而且在容器部署環境下，單 Consumer 不能起太多執行緒去支撐高併發；在容器這個環境下，比較好的方式是多例項，然後按照 CPU 的核心數，啟動多個執行緒，比如 8C 的啟動 8 個執行緒，因為容器是有 Quota 的，一般是 1C,2C,4C,8C 這樣，這種情況下，如果執行緒數超過了 CPU 的核心數，其實對併發度並沒有太大的意義。

接下來，分享一下做這個接入方式的時候遇到的一些問題，如下圖所示：

1、訊息大小的限制。

因為這裡有一層 RPC ，在 RPC 請求過程中會有單次請求大小的限制；另外一方面是 RocketMQ 的 producer 裡會有一個 MaxMessageSize 方法去控制訊息不能超過這個大小； Broker 裡也有一個引數，是 Broker 啟動的配置，這個需要Broker重啟，不然修改也不生效， Broker 裡面有一個 DefaultAppendMessage 配置，是在啟動的時候傳進去對的引數，如果僅 NameServer 線上變更是不生效的，而且超過這個大小會報錯。因為現在 RocketMQ 預設是 4M 的訊息，如果將 RocketMQ 作為日誌匯流排，可能訊息體大小不是太夠， Procuer 和 Broker 是都需要做變更的。

2、多連線的問題。

如果看 RocketMQ 原始碼會發現，多個 Producer 是共享一個底層的 MQ Client 例項的，因為一個 socket 連線吞吐是有限的，所以只會和Broker建立一個socket連線。另外，我們也有 socket 與 socket 之間是隔離的，可以通過 Producer 的 setIntanceName() ，當與 DefaultI Instance 的 name 不一樣時會新啟動一個 Client 的，其實就是一個新的 socket 連線，對於有隔離需求的、連線池需求得等，這個引數是有用的，在 4.5.0 上新加了一個介面是指定構造的例項數量。

3、超時設定。

因為多了一層 RPC ，那一層是有一個超時設定的，這個會有點不一樣，因為我們的 RPC 請求裡會帶上超時設定的，客戶端到 Proxy 有一個 RTT ，然後 Producer 到 Broker 的傳送訊息也是有一個請求響應延時，需要給 SDK 一個正確的超時語義。

4、如何選擇一個合適的 reblance 演算法，我們遇到這個問題是在雙機房同城容災的背景下，會有一邊 Topic 的 MessageQueue 沒有寫入。

這種情況下， RocketMQ 自己預設的是按照平均分配演算法進行分配的，比如有 10 個 Queue ， 3 個 Proxy 情況， 1、2、3 是對應 Proxy1，4、5、6 是對應 Proxy2，7、8、9、10 是對應 Proxy3 ，如果在雙機房同城容災部署情況下，一般有一半 Message Queue 是沒有寫入的，會有一大部分 Consumer 是啟動了，但是分配到的 Message Queue 是沒有訊息寫入的。然後另外一個訴求是因為有跨機房的流量，所以他其實直接複用開源出來的 Consumer 的實現裡就有根據 MachineRoom 去做 reblance ，會就近分配你的 MessageQueue 。

5、在 Proxy 這邊需要做一個快取，特別是拉訊息的快取。

特別提醒一下， Proxy 拉訊息都是通過 Slave 去拉，不需要使用 Master 去拉， Master 的 IO 比較重；還有 Buffer 的管理，我們是遇到過這種問題的，如果只考慮 Message 數量的話，會導致 OOM ，所以要注意訊息 size 的設定，

6、端到端壓縮。

因為 RocketMQ 在訊息超過 4k 的時候， Producer 會進行壓縮。如果不在客戶端做壓縮，這還是涉及到 RPC 的問題， RPC 一般來說， Byte 型別，就是 Byte 陣列型別它是不會進行壓縮的，只是會進行一些常規的編碼，所以訊息體需要在客戶端做壓縮。如果放在 Proxy 這邊做， Proxy 壓力會比較大，所以不如放在客戶端去承載這個壓縮。

頭條的容災系統建設

前面大致介紹了我們這邊大致如何接入 RocketMQ ，如何實現這麼一套 Proxy ，以及在實現這套 Proxy 過程中遇到的一些問題。下面看一下災難恢復的方案，設計之初也參考了一些潛在相關方案。

第一種方案：擴充套件叢集，擴充套件叢集的方案就像下圖所示。

這是 master 和 slave 跨機房去部署的方式。因為我們有一層 proxy ，所以可以很方便的去做流量的排程，讓訊息只在一個主機房進行訊息寫入，不需要一個類似中控功能的實體存在。

第二種方案：類似 MySQL 和 Redis 的架構模式，即單主模式，只有一個地方式寫入的，如下圖所示。資料是通過 Mysql Matser/Slave 方式同步到另一個機房。這樣 RocketMQ 會啟動一個類似 Kafka 的 Mirror maker 類進行訊息複製，這樣會多一倍的冗餘，實際上資料還會存在一些不一致的問題。

第三種方案：雙寫加雙向複製的架構。這個結構太複雜不好控制，尤其是雙向複製，其中訊息區迴環的問題比較好解決，只需針對在每個正常的業務訊息，在 Header 里加一個標誌欄位就好，另外的 Mirror 發現有這個欄位就把這條訊息直接丟掉即可。這個鏈路上維護複雜而且存在資料冗餘，其中最大問題是兩邊的資料不對等，在一邊掛掉情況下，對於一些無法接受資料不一致的是有問題的。

此外，雙寫都是沒有 Mirror 的方案，如下圖所示。這也是我們最終選擇的方案。我們對有序訊息和無序訊息的處理方式不太一樣，針對無序訊息只需就近寫本機房就可以了，對於有序訊息我們還是會有一個主機房，Proxy 會去 NameServer 拉取 Broker 的 Queue 資訊， Producer 將有序訊息路由到一個指定主機房，消費端這一側，就是就近拉取訊息。對於順序訊息我們會採取一定的排程邏輯保證均衡的分擔壓力獲取訊息，這個架構的優點是比較簡單，缺點是當叢集中一邊掛掉時，會造成有序訊息的無序，這邊是通過記錄訊息 offset 來處理的。

此外，還有一種獨立叢集部署的，相當於沒有上圖中間的有序訊息那條線，因為大多數有序訊息是整體體系的，服務要部署單元化，比如某些 uid 、訂單 Id 的訊息或請求只會落到一邊機房的，完全不用擔心訊息來得時候是否需要按照某些 key 去指定 MessageQueue ,因為過來的訊息必定是隸屬於這個機房的，也就是說中間有序訊息那條線可以不用關心了，可以直接去掉。但是，這個是和整個公司部署方式以及單元化體系有關係的，對於部分業務我們是直接做到兩個叢集，兩邊的生產者、消費者、Broker 、Proxy 全部是隔離的，兩邊都互不發現，就是這麼一套執行方式，但是這就需要業務的上下游要做到單元化的程度才可行。