本篇文章整理自今日頭條的沈輝在 RocketMQ 開發者沙龍中的演講，主要和大家分享一下，RocketMQ 在微服務架構下的實踐和容災體系建設。沈輝是今日頭條的架構師，主要負責 RocketMQ 在頭條的落地以及架構設計，參與消息系統的時間大概一年左右。

以下是本次分享的議題：

• 頭條的業務背景

• 為什么選擇 RocketMQ

• RocketMQ 在頭條的落地實踐

• 頭條的容災系統建設

業務背景

今日頭條的服務大量使用微服務，容器數目巨大，業務線繁多， 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 全部是隔離的，兩邊都互不發現，就是這么一套運行方式，但是這就需要業務的上下游要做到單元化的程度才可行。

以上就是 RocketMQ 在頭條的落地實踐頭條的容災系統建設分享，謝謝。

作者信息：沈輝，畢業于北京郵電大學，就職于字節跳動基礎架構，主要參與負責消息隊列服務的開發與維護。

原文鏈接

本文為云棲社區原創內容，未經允許不得轉載