簡介：快手需要建設(shè)一個主要面向在線業(yè)務(wù)的消息系統(tǒng)作為 Kafka 的補充，低延遲、高并發(fā)、高可用、高可靠的分布式消息中間件 RocketMQ 正是我們所需的。

作者：黃理

黃理，10多年軟件開發(fā)和架構(gòu)經(jīng)驗，熱衷于代碼和性能優(yōu)化，開發(fā)和參與過多個開源項目。曾在淘寶任業(yè)務(wù)架構(gòu)師多年，當前在快手負責在線消息系統(tǒng)建設(shè)工作。

?

為什么建設(shè)在線消息系統(tǒng)

在引入RocketMQ之前，快手已經(jīng)在大量的使用Kafka了，但并非所有情況下Kafka都是最合適的，比如以下場景：

• 業(yè)務(wù)希望個別消費失敗以后可以重試，并且不堵塞后續(xù)其它消息的消費。
• 業(yè)務(wù)希望消息可以延遲一段時間再投遞。
• 業(yè)務(wù)需要發(fā)送的時候保證數(shù)據(jù)庫操作和消息發(fā)送是一致的（也就是事務(wù)發(fā)送）。
• 為了排查問題，有的時候業(yè)務(wù)需要一定的單個消息查詢能力。

為了應(yīng)對以上這類場景，我們需要建設(shè)一個主要面向在線業(yè)務(wù)的消息系統(tǒng)，作為Kafka的補充。在考察的一些消息中間件中，RocketMQ和業(yè)務(wù)需求匹配度比較高，同時部署結(jié)構(gòu)簡單，使用的公司也比較多，于是最后我們就采用了RocketMQ。

部署模式和落地策略

在一個已有的體系內(nèi)落地一個開軟軟件，通常大概有兩種方式：

?

方式一，在開源軟件的基礎(chǔ)上做深度修改，很容易實現(xiàn)公司內(nèi)需要的定制功能。但和社區(qū)開源版本分道揚鑣，以后如何升級？

方式二，盡量不修改社區(qū)版本（或減少不兼容的修改），而是在它的外圍或者上層進一步包裝來實現(xiàn)公司內(nèi)部需要的定制功能。

?

注：上圖方式一的圖畫的比較極端，實際上很多公司是方式一、方式二結(jié)合的。

?

我們選擇了方式二。最早的時候，我們使用的是4.5.2版本，后來社區(qū)4.7版本大幅減小了同步復制的延遲，正好我們的部署模式就是同步復制，于是就很輕松的升級了4.7系列，享受了新版本的紅利。

?

在部署集群的時候，還會面臨很多部署策略的選擇：

???????大集群 vs 小集群

???????選擇副本數(shù)

???????同步刷盤 vs 異步刷盤

???????同步復制? vs 異步復制

???????SSD vs 機械硬盤

大集群會有更好的性能彈性，而小集群具有更好的隔離型，此外小集群可以不需要跨可用區(qū)/IDC部署，所以會有更好的健壯性，我們非常看重穩(wěn)定性，因此選擇了小集群。集群同步復制異步刷盤，首選SSD。

客戶端封裝策略

如上所述，我們沒有在Rocketmq里面做深度修改，所以需要提供一個SDK來提供公司內(nèi)的需要的定制功能，這個SDK大概是這樣的：

??

對外只提供最基本的API，所有訪問必須經(jīng)過我們提供的接口。簡潔的API就像冰山的一個角，除了對外的簡單接口，下面所有的東西都可以升級更換，而不會破壞兼容性。

?

業(yè)務(wù)開發(fā)起來也很簡單，只要需要提供Topic（全局唯一）和Group就可以生產(chǎn)和消費，不用提供環(huán)境、Name Server地址等。SDK內(nèi)部會根據(jù)Topic解析出集群Name Server的地址，然后連接相應(yīng)的集群。生產(chǎn)環(huán)境和測試環(huán)境環(huán)境會解析出不同的地址，從而實現(xiàn)了隔離。

?

上圖分為3層，第二層是通用的，第三層才對應(yīng)具體的MQ實現(xiàn)，因此，理論上可以更換為其它消息中間件，而客戶端程序不需要修改。

?

SDK內(nèi)部集成了熱變更機制，可以在不重啟client的情況下做動態(tài)配置，比如下發(fā)路由策略（更換集群name server的地址，或者連接到別的集群去），Client的線程數(shù)、超時時間等。通過maven強制更新機制，可以保證業(yè)務(wù)使用的SDK基本上是最新的。

集群負載均衡 & 機房災(zāi)備

所有的Topic默認都分配到兩個可用區(qū)，生產(chǎn)者和消費者會同時連接至少兩個獨立集群（分布在不同的可用區(qū)），如下圖：

?

生產(chǎn)者同時連接兩個集群，如果可用區(qū)A出現(xiàn)故障，流量就會自動切換到可用區(qū)B的集群2去。我們開發(fā)了一個小組件來實現(xiàn)自適應(yīng)的集群負載均衡，它包含以下能力：

???????千萬級OPS

???????靈活的權(quán)重調(diào)整策略

???????健康檢查支持/事件通知

???????并發(fā)度控制（自動降低響應(yīng)慢的服務(wù)器的請求數(shù)）

???????資源優(yōu)先級（類似Envoy，實現(xiàn)本地機房優(yōu)先，或是被調(diào)服務(wù)器很多的時候選取一個子集來調(diào)用）

???????自動優(yōu)先級管理

???????增量熱變更

實際上它并不僅僅用于消息生產(chǎn)者，而是一個通用的主調(diào)方負載均衡類庫，可以在github上找到：

https://github.com/PhantomThief/simple-failover-java

核心的SimpleFailover接口和PriorityFailover類沒有傳遞第三方依賴，非常容易整合。

多樣的消息功能

延遲消息

延遲消息是非常重要的業(yè)務(wù)功能，不過RocketMQ內(nèi)置的延遲消息只能支持幾個固定的延遲級別，所以我們又開發(fā)了單獨的Delay Server來調(diào)度延遲消息：

上圖這個結(jié)構(gòu)沒有直接將延遲消息發(fā)到Delay Server，而是更換Topic以后存入RocketMQ。這樣的好處是可以復用現(xiàn)有的消息發(fā)送接口（以及上面的所有擴展能力）。對業(yè)務(wù)來說，只需要在構(gòu)造消息的時候額外指定一個延遲時間字段即可，其它用法都不變。

事務(wù)消息

RocketMQ 4.3版本以后支持了事務(wù)消息，可以保證本地事務(wù)和消費發(fā)送同時成功或者失敗，對于一些業(yè)務(wù)場景很有幫助。事務(wù)消息的用法和原理有很多資料，這里就不細述了。但關(guān)于事務(wù)消息的實踐網(wǎng)上資料較少，我們可以給出一些建議。

?

首先，事務(wù)消息功能一直在不斷完善，應(yīng)該使用最新的版本，至少是4.6.1以后的版本，可以避免很多問題。

?

其次，事務(wù)消息性能是不如普通消息的，它在內(nèi)部實際上會生成3個消息（一階段1個，二階段2個），所以性能大約只有普通消息的1/3，如果事務(wù)消息量大的話，要做好容量規(guī)劃。回查調(diào)度線程也只有1個，不要用極限壓力去考驗它。

?

最后有一些參數(shù)注意事項。在broker的配置中：

• transientStorePoolEnable這個參數(shù)必須保持默認值false，否則會有嚴重的問題。
• endTransactionThreadPoolNums是事務(wù)消息二階段處理線程大小，sendMessageThreadPoolNums則指定一階段處理線程池大小。如果二階段的處理速度跟不上一階段，就會造成二階段消息丟失導致大量回查，所以建議endTransactionThreadPoolNums應(yīng)該大于sendMessageThreadPoolNums，建議至少4倍。
• useReentrantLockWhenPutMessage設(shè)置為true（默認值是false），以免線程搶鎖出現(xiàn)嚴重的不公平，導致二階段處理線程長時間搶不到鎖。
• transactionTimeOut默認值6秒太短了，如果事務(wù)執(zhí)行時間超過6秒，就可能導致消息丟失。建議改到1分鐘左右。

?

生產(chǎn)者client也有一個注意事項，如果有多組broker，并且是2副本（有1個Slave），應(yīng)該打開retryAnotherBrokerWhenNotStoreOK，以免某個Slave出現(xiàn)故障以后，大量消息發(fā)送失敗。

?

分布式對賬監(jiān)控

除了比較一些常規(guī)的監(jiān)控手段以外，我們開發(fā)了一個監(jiān)控程序做分布式對賬。可以發(fā)現(xiàn)我們的集群以及我們提供的SDK是否有異常。

?

?

具體做法是在每個Broker上都建立一個監(jiān)控專用的Topic，監(jiān)控程序使用我們自己提供的SDK框架來連接集群（就像我們的業(yè)務(wù)用戶那樣），監(jiān)控生產(chǎn)者會給每個集群發(fā)送少量消息。然后檢查發(fā)送是否成功：

發(fā)送成功	成功
	刷盤超時
	Slave超時
	Slave不可用
發(fā)送失敗	具體錯誤碼

生產(chǎn)者只對這些結(jié)果進行打點，不判斷是否正常，具體到監(jiān)控（或者演練）場景可以配置不同的報警規(guī)則。

?

消費者收到了消息會通過TCP旁路Ack生產(chǎn)者，生產(chǎn)者這邊會做分布式對賬，將對賬結(jié)果打點：

• 收到消息
• 消息丟失（或超時未收到消息）
• 重復收到消息
• 消息生成到最終消費的時間差
• Ack生產(chǎn)者失敗（由消費者打點）

同樣監(jiān)控程序只負責打點，報警規(guī)則可另外配置。

?

這套機制也可以用于分布式性能壓測和故障演練。在做壓測的時候，每個消息都Ack的話，對生產(chǎn)者的內(nèi)存壓力很大，因為它發(fā)出去的消息，需要在內(nèi)存中保留一段時間（直到到達這個消息的對賬時間），這段時間消費者Ack或者重復Ack都需要記錄。所以我們實現(xiàn)了按比例抽樣對賬的功能，開啟以后只有需要對賬的消息才會在內(nèi)存中保留一段時間。

?

順便說一下，我們做壓測時，合格的標準是異步生產(chǎn)不失敗、消費不延遲、每一個消息都不丟失。這樣做是為了保證壓測時能給出更加準確的，可供線上系統(tǒng)參考的性能數(shù)字，而不是制造理想條件，追求一個大的數(shù)字。比如異步生產(chǎn)比同步生產(chǎn)更脆弱（壓測client如果同步生產(chǎn)，broker抖動的時候，同步client會被堵塞導致發(fā)送速度降低，于是降低了broker壓力，消息發(fā)送不容易失敗，但是會看到發(fā)送速率在波動），更貼近生產(chǎn)環(huán)境的實際情況，我們就選擇異步生產(chǎn)來評估。

?

性能優(yōu)化

Broker默認的參數(shù)在我們的場景下（SSD、同步復制、異步刷盤）不是最優(yōu)的，有的參數(shù)也許在大多數(shù)場景下都不是最優(yōu)的。我們列出一些重要的參數(shù)，供大家參考：

?

參數(shù)	默認值	說明
flushCommitLogTimed	False	默認值不合理，異步刷盤這個參數(shù)應(yīng)該設(shè)置成true，導致頻繁刷盤，對性能影響極大
deleteWhen	04	幾點刪除過期文件的時間，刪除文件時有很多磁盤讀，這個默認值是合理的，有條件的話還是建議低峰刪除
sendMessageThreadPoolNums	1	處理生產(chǎn)消息的線程數(shù)，這個線程干的事情很多，建議設(shè)置為2～4，但太多也沒有什么用。因為最終寫commit log的時候只有一個線程能拿到鎖。
useReentrantLockWhenPutMessage	False	如果前一個參數(shù)設(shè)置比較大，這個最好設(shè)置為True，避免高負載下自旋鎖空轉(zhuǎn)消耗CPU。
sendThreadPoolQueueCapacity	10000	處理生產(chǎn)消息的隊列大小，默認值可能有點小，比如5萬TPS（異步發(fā)送）的情況下，卡200ms就會爆。設(shè)置比較小的數(shù)字可能是擔心有大量大消息撐爆內(nèi)存（比如100K的話，1萬個的消息大概占用1G內(nèi)存，也還好），具體可以自己算，如果都是小消息，可以把這個數(shù)字改大。可以修改broker參數(shù)限制client發(fā)送大消息。
brokerFastFailureEnable	True	Broker端快速失敗（限流），和下面兩個參數(shù)配合。這個機制可能有爭議，client設(shè)置了超時時間，如果client還愿意等，并且sendThreadPoolQueue還沒有滿，不應(yīng)該失敗，sendThreadPoolQueue滿了自然會拒絕新的請求。但如果client設(shè)置的超時時間很短，沒有這個機制可能導致消息重復。可以自行決定是否開啟。理想情況下，能根據(jù)client設(shè)置的超時時間來清理隊列是最好的。
waitTimeMillsInSendQueue	200	200ms很容易導致發(fā)送失敗，建議改大，比如1000
osPageCacheBusyTimeOutMills	1000	Page cache超時時間，如果內(nèi)存比較多，比如32G以上，建議改大點

?

?

得益于簡單、幾乎0依賴的部署模式，使得我們部署小集群的成本非常低；不對社區(qū)版本進行魔改，保證我們可以及時升級；統(tǒng)一SDK入口方便集群維護和功能升級；通過復合小集群+自動負載均衡實現(xiàn)多機房多活；充分利用RocketMQ的功能比如事務(wù)消息、延遲消息（增強）來滿足業(yè)務(wù)的多樣性需求；通過自動的分布式對賬，對每一個Broker以及我們的SDK進行正確性監(jiān)控；本問也進行了一些性能參數(shù)的分享，但寫的比較簡單，基本只說了怎么調(diào)，但沒能細說為什么，以后我們會另寫文章詳述。目前RocketMQ已經(jīng)應(yīng)用在公司在大多數(shù)業(yè)務(wù)線，期待將來會有更好的發(fā)展！

?

掃碼了解更多技術(shù)干貨與客戶案例：

---

原文鏈接：https://developer.aliyun.com/article/782584?

版權(quán)聲明：本文內(nèi)容由阿里云實名注冊用戶自發(fā)貢獻，版權(quán)歸原作者所有，阿里云開發(fā)者社區(qū)不擁有其著作權(quán)，亦不承擔相應(yīng)法律責任。具體規(guī)則請查看《阿里云開發(fā)者社區(qū)用戶服務(wù)協(xié)議》和《阿里云開發(fā)者社區(qū)知識產(chǎn)權(quán)保護指引》。如果您發(fā)現(xiàn)本社區(qū)中有涉嫌抄襲的內(nèi)容，填寫侵權(quán)投訴表單進行舉報，一經(jīng)查實，本社區(qū)將立刻刪除涉嫌侵權(quán)內(nèi)容。

總結(jié)

以上是生活随笔為你收集整理的快手基于RocketMQ的在线消息系统建设实践的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。