同程藝龍的機(jī)票、火車票、汽車票、酒店相關(guān)業(yè)務(wù)已經(jīng)接入了 RocketMQ，用于流量高峰時(shí)候的削峰，以減少后端的壓力。

同時(shí)，對(duì)常規(guī)的系統(tǒng)進(jìn)行解耦，將一些同步處理改成異步處理，每天處理的數(shù)據(jù)達(dá) 1500 億條。

在近期的 Apache RocketMQ Meetup 上，同程藝龍機(jī)票事業(yè)部架構(gòu)師查江，分享了同程藝龍的消息系統(tǒng)如何應(yīng)對(duì)每天 1500 億條的數(shù)據(jù)處理。

通過(guò)此文，您將了解到：

• 同程藝龍消息系統(tǒng)的使用情況
• 同程藝龍消息系統(tǒng)的應(yīng)用場(chǎng)景
• 技術(shù)上踩過(guò)的坑
• 基于 RocketMQ 的改進(jìn)

同程藝龍消息系統(tǒng)的使用情況

RocketMQ 集群分為 Name Server 和 Broker 兩部分，Name Server 用的是雙主模式，一個(gè)是考慮性能，另一個(gè)考慮安全性。在純數(shù)據(jù)的 Broker 分成很多組，每個(gè)組里面分為 Master 和 Slave。

目前，我們的機(jī)票、火車票、汽車票、酒店相關(guān)業(yè)務(wù)已經(jīng)接入了 RocketMQ，用于流量高峰時(shí)候的削峰，以減少后端的壓力。

同時(shí)，對(duì)常規(guī)的系統(tǒng)進(jìn)行解耦，將一些同步處理改成異步處理，每天處理的數(shù)據(jù)達(dá) 1500 億條。

選擇 RocketMQ 的原因是：

• 接入簡(jiǎn)單，引入的 Java 包比較少
• 純 Java 開(kāi)發(fā)，設(shè)計(jì)邏輯比較清晰
• 整體性能比較穩(wěn)定的，Topic 數(shù)量大的情況下，可以保持性能

同程藝龍消息系統(tǒng)的應(yīng)用場(chǎng)景

退訂系統(tǒng)

這個(gè)是我們退訂系統(tǒng)中的一個(gè)應(yīng)用場(chǎng)景。用戶點(diǎn)擊前端的退訂按鈕，系統(tǒng)調(diào)用退訂接口，再去調(diào)用供應(yīng)商的退訂接口，從而完成一個(gè)退訂功能。

如果供應(yīng)商的系統(tǒng)接口不可靠，就會(huì)導(dǎo)致用戶退訂失敗，如果系統(tǒng)設(shè)置為同步操作，會(huì)導(dǎo)致用戶需要再去點(diǎn)一次。

所以，我們引入 RocketMQ 將同步改為異步，當(dāng)前端用戶發(fā)出退訂需求，退訂系統(tǒng)接收到請(qǐng)求后就會(huì)記錄到退訂系統(tǒng)的數(shù)據(jù)庫(kù)里面，表示這個(gè)用戶正在退訂。

同時(shí)通過(guò)消息引擎把這條退訂消息發(fā)送到和供應(yīng)商對(duì)接的系統(tǒng)，去調(diào)用供應(yīng)商的接口。

如果調(diào)用成功，就會(huì)把數(shù)據(jù)庫(kù)進(jìn)行標(biāo)識(shí)，表示已經(jīng)退訂成功。同時(shí)，加了一個(gè)補(bǔ)償?shù)哪_本，去數(shù)據(jù)庫(kù)撈那些未退訂成功的消息，重新退訂，避免消息丟失引起的退訂失敗的情況。

房倉(cāng)系統(tǒng)

第二個(gè)應(yīng)用場(chǎng)景是我們的房倉(cāng)系統(tǒng)。這是一個(gè)比較常規(guī)的消息使用場(chǎng)景，我們從供應(yīng)商處采集一些酒店的基本信息數(shù)據(jù)和詳情數(shù)據(jù)，然后接入到消息系統(tǒng)，由后端的分銷系統(tǒng)、最小價(jià)系統(tǒng)和庫(kù)存系統(tǒng)來(lái)進(jìn)行計(jì)算。

同時(shí)當(dāng)供應(yīng)商有變價(jià)的時(shí)候,變價(jià)事件也會(huì)通過(guò)消息系統(tǒng)傳遞給我們的后端業(yè)務(wù)系統(tǒng)，來(lái)保證數(shù)據(jù)的實(shí)時(shí)性和準(zhǔn)確性。

供應(yīng)庫(kù)的訂閱系統(tǒng)

數(shù)據(jù)庫(kù)的訂閱系統(tǒng)也用到了消息的應(yīng)用。一般情況下做數(shù)據(jù)庫(kù)同步，都是通過(guò) binlog 去讀里面的數(shù)據(jù)，然后搬運(yùn)到數(shù)據(jù)庫(kù)。

搬運(yùn)過(guò)程中，我們最關(guān)注的是數(shù)據(jù)的順序性，因此在數(shù)據(jù)庫(kù) row 模式的基礎(chǔ)上，新增了一個(gè)功能，以確保每一個(gè) Queue 里面的順序是唯一的。

雖然 Queue 里面的順序天然都是唯一的，但我們?cè)谑褂蒙嫌幸粋€(gè)特點(diǎn)，就是把相同 ID 的消息都是放在同一個(gè) Queue 里面的。

例如，圖中右上角 id1 的消息，數(shù)據(jù)庫(kù)主字段是 id1，就統(tǒng)一放在 Queue1 里面，而且是順序的。

在 Queue2 里，兩個(gè) id3 之間被兩個(gè)順序的 id2 間隔開(kāi)來(lái)了，但實(shí)際消費(fèi)讀出來(lái)的時(shí)候，也會(huì)是順序的，由此，可以用多隊(duì)列的順序來(lái)提高整體的并發(fā)度。

技術(shù)上踩過(guò)的坑

供應(yīng)商系統(tǒng)的場(chǎng)景

上圖中，一個(gè) MQ 對(duì)應(yīng)有兩個(gè)消費(fèi)者，他們是在同一個(gè) Group1 中，起初大家都只有 Topic1，這時(shí)候是正常消費(fèi)的。

但如果在***個(gè)消費(fèi)者里面加入一個(gè) Topic2，這時(shí)候是無(wú)法消費(fèi)或消費(fèi)不正常了。

這是 RocketMQ 本身的機(jī)制引起的問(wèn)題，需要在第二個(gè)消費(fèi)者里面加入 Topic2 才能正常消費(fèi)。

支付交易系統(tǒng)的場(chǎng)景

另外一個(gè)是支付交易系統(tǒng)，這個(gè)場(chǎng)景下也是有兩個(gè)應(yīng)用，他們都是在同一 Group 和同一 Topic 下，一個(gè)是消費(fèi) Tag1 的數(shù)據(jù)，另一個(gè)是消費(fèi) Tag2 的數(shù)據(jù)。

正常情況下，啟動(dòng)應(yīng)該是沒(méi)問(wèn)題的，但是有一天我們發(fā)現(xiàn)一個(gè)應(yīng)用起不來(lái)了，另外一個(gè)應(yīng)用，他只消費(fèi) Tag2 的數(shù)據(jù)，但是因?yàn)?RocketMQ 的機(jī)制會(huì)把 Tag1 的數(shù)據(jù)拿過(guò)來(lái)，拿過(guò)來(lái)后會(huì)把 Tag1 的數(shù)據(jù)丟棄。

這會(huì)導(dǎo)致用戶在支付過(guò)程中出現(xiàn)支付失敗的情況。對(duì)此，我們把 Tag2 放到 Group2 里面，兩個(gè) Group 就不會(huì)消費(fèi)相同的消息了。

個(gè)人建議 RocketMQ 能夠?qū)崿F(xiàn)一個(gè)機(jī)制，即只接受自己的 Tag 消息，非相關(guān)的 Tag 不去接收。

大量老數(shù)據(jù)讀取的場(chǎng)景

在火車票消費(fèi)的場(chǎng)景中，我們發(fā)現(xiàn)有 200 億條老數(shù)據(jù)沒(méi)有被消費(fèi)。當(dāng)我們消費(fèi)啟動(dòng)的時(shí)候，RocketMQ 會(huì)默認(rèn)從第 0 個(gè)數(shù)據(jù)開(kāi)始讀，這時(shí)候磁盤 IO 飆升到 100%，從而影響到其他消費(fèi)端數(shù)據(jù)的讀取，但這些老數(shù)據(jù)被加載后，并沒(méi)有實(shí)際作用。

因此，對(duì)于大量老數(shù)據(jù)讀取的改進(jìn)方式是：

• 對(duì)于新消費(fèi)組，默認(rèn)從 LAST_OFFSET 消費(fèi)。
• Broker 中單 Topic 堆積超過(guò) 1000 萬(wàn)時(shí)，禁止消費(fèi)，需聯(lián)系管理員開(kāi)啟消費(fèi)。
• 監(jiān)控要到位，磁盤 IO 飆升時(shí)，能立刻聯(lián)系到消費(fèi)方處理。

服務(wù)端的場(chǎng)景

①CentOS 6.6 中 Futex Kernel bug, 導(dǎo)致 Name Server, Broker 進(jìn)程經(jīng)常掛起，無(wú)法正常工作

解決方法：升級(jí)到 6.7

②服務(wù)端 2 個(gè)線程會(huì)創(chuàng)建相同 CommitLog 放入 List，導(dǎo)致計(jì)算消息 offset 錯(cuò)誤，解析消息失敗，無(wú)法消費(fèi)，重啟沒(méi)法解決問(wèn)題。

解決方法：線程安全問(wèn)題，改為單線程

③Pull 模式下重置消費(fèi)進(jìn)度，導(dǎo)致服務(wù)端填充大量數(shù)據(jù)到 Map 中，Broker CPU 使用率飆升 100%。

解決方法：Map 局部變量場(chǎng)景用不到，刪除

④Master 建議客戶端到 Slave 消費(fèi)時(shí)，若數(shù)據(jù)還沒(méi)同步到 Slave, 會(huì)重置 pullOffset，導(dǎo)致大量重復(fù)消費(fèi)。

解決方法：不重置 offset

⑥同步?jīng)]有 MagicCode，安全組掃描同步端口時(shí)，Master 解析錯(cuò)誤，導(dǎo)致一些問(wèn)題。

解決方法：同步時(shí)添加 magicCode 校驗(yàn)

基于 RocketMQ 的改進(jìn)

新增客戶端

新增 .net 客戶端，基于 Java 源代碼原生開(kāi)發(fā);新增 HTTP 客戶端，實(shí)現(xiàn)部分功能，并通過(guò) Netty Server 連接 RocketMQ。

新增消息限流功能

如果客戶端代碼寫錯(cuò)產(chǎn)生死循環(huán)，就會(huì)產(chǎn)生大量的重復(fù)數(shù)據(jù)，這時(shí)候會(huì)把生產(chǎn)線程打滿，導(dǎo)致隊(duì)列溢出，嚴(yán)重影響我們 MQ 集群的穩(wěn)定性，從而影響其他業(yè)務(wù)。

上圖是限流的模型圖，我們把限流功能加在 Topic 之前。通過(guò)限流功能可以設(shè)置 rate limit 和 size limit 等。

其中 rate limit 是通過(guò)令牌桶算法來(lái)實(shí)現(xiàn)的，即每秒往桶里放多少個(gè)令牌，每秒就消費(fèi)多少速度，或者是往 Topic 里寫多少數(shù)據(jù)。以上的兩個(gè)配置是支持動(dòng)態(tài)修改的。

后臺(tái)監(jiān)控

我們還做了一個(gè)監(jiān)控后臺(tái)，用于監(jiān)控消息的全鏈路過(guò)程，包括：

• 消息全鏈路追蹤，覆蓋消息產(chǎn)生、消費(fèi)、過(guò)期整個(gè)生命周期
• 消息生產(chǎn)、消費(fèi)曲線
• 消息生產(chǎn)異常報(bào)警
• 消息堆積報(bào)警，通知哪個(gè) IP 消費(fèi)過(guò)慢

其他功能：

• HTTP 方式生產(chǎn)，消費(fèi)消息
• Topic 消費(fèi)權(quán)限設(shè)置，Topic 只能被指定 Group 消費(fèi)，防止線上錯(cuò)亂訂閱
• 支持新消費(fèi)組從***位置消費(fèi) (默認(rèn)是從***條開(kāi)始消費(fèi))
• 廣播模式消費(fèi)進(jìn)度同步 (服務(wù)端顯示進(jìn)度)

以上便是同程藝龍?jiān)谙⑾到y(tǒng)建設(shè)方面的實(shí)踐。