前言

在之前的文章中，我們提到車聯(lián)網(wǎng) TSP 平臺擁有很多不同業(yè)務(wù)的主題，并介紹了 ??如何根據(jù)不同業(yè)務(wù)場景進行 MQTT 主題設(shè)計??。車輛會持續(xù)不斷產(chǎn)生海量的消息，每一條通過車聯(lián)網(wǎng)上報的數(shù)據(jù)都是非常珍貴的，其背后蘊藏著巨大的業(yè)務(wù)價值。因此我們構(gòu)建的車輛 TSP 平臺也通常需要擁有千萬級主題和百萬級消息吞吐能力。

傳統(tǒng)的互聯(lián)網(wǎng)系統(tǒng)很難支撐百萬量級的消息吞吐。在本文中，我們將主要介紹如何針對百萬級消息吞吐這一需求進行新一代車聯(lián)網(wǎng)平臺架構(gòu)設(shè)計。

車聯(lián)網(wǎng)場景消息吞吐設(shè)計的關(guān)聯(lián)因素

車聯(lián)網(wǎng)的消息分為上行和下行。上行消息一般是傳感器及車輛發(fā)出的告警等消息，把設(shè)備的信息發(fā)送給云端的消息平臺。下行消息一般有遠程控制指令集消息和消息推送，是由云端平臺給車輛發(fā)送相應(yīng)的指令。

在車聯(lián)網(wǎng)消息吞吐設(shè)計中，我們需要重點考慮以下因素：

(1) 消息頻率

車在行駛過程中，GPS、車載傳感器等一直不停地在收集消息，為了收到實時的反饋信息，其上報接收的消息也是非常頻繁的。上報頻率一般在 100ms-30s 不等，所以當車輛數(shù)量達到百萬量級時，平臺就需要支持每秒百萬級的消息吞吐。

(2) 消息包大小

消息是通過各種傳感器來采集自身環(huán)境和狀態(tài)信息(車聯(lián)網(wǎng)場景常見的有新能源國標數(shù)據(jù)和企標數(shù)據(jù))。整個消息包大小一般在 500B 到幾十 KB 不等。當大量消息包同時上報時，需要車聯(lián)網(wǎng)平臺擁有更強的接收、發(fā)送大消息包的能力。

(3) 消息延時

車輛在行駛過程中，消息數(shù)據(jù)只能通過無線網(wǎng)絡(luò)來進行傳輸。在大部分車聯(lián)網(wǎng)場景下，對車輛的時延要求是 ms 級別。平臺在滿足百萬級吞吐條件下，還需要保持低延時的消息傳輸。

(4) Topic 數(shù)量和層級

在考慮百萬級消息吞吐場景時，還需要針對消息 Topic 數(shù)量和 Topic 樹層級進行規(guī)范設(shè)計。

(5) Payload 編解碼

當消息包比較大的時候，需要重點考慮消息體的封裝。單純的 JSON 封裝在消息解析時不夠高效，可以考慮采用 Avro、Protobuf 等編碼格式進行 Payload 格式化封裝。

對于百萬級消息吞吐場景，基于 MQTT 客戶端共享訂閱消息或通過規(guī)則引擎實時寫入關(guān)系型數(shù)據(jù)庫的傳統(tǒng)架構(gòu)顯然無法滿足。目前主流的架構(gòu)選型有兩種：一種是消息接入產(chǎn)品/服務(wù)+消息隊列(Kafka、Pulsar、RabbitMQ、RocketMQ 等)，另外一種是消息接入產(chǎn)品/服務(wù)+時序數(shù)據(jù)庫(InfluxDB、TDengine、Lindorm等)來實現(xiàn)。

接下來我們將基于上述的關(guān)聯(lián)因素和客戶案例的最佳實踐，以云原生分布式物聯(lián)網(wǎng)消息服務(wù)器 EMQX 作為消息接入層，分別介紹這兩種架構(gòu)的實現(xiàn)方式。

EMQX+Kafka 構(gòu)建百萬級吞吐車聯(lián)網(wǎng)平臺

架構(gòu)設(shè)計

Kafka 作為主流消息隊列之一，具有持久化數(shù)據(jù)存儲能力，可進行持久化操作，同時可通過將數(shù)據(jù)持久化到硬盤以及 replication 防止數(shù)據(jù)丟失。后端 TSP 平臺或者大數(shù)據(jù)平臺可以批量訂閱想要的消息。

由于 Kafka 擁有訂閱發(fā)布的能力，既可以從南向接收，把上報消息緩存起來;又可以通過北向的連接，把需要發(fā)送的指令通過接口傳輸給前端，用作指令下發(fā)。

我們以 Kafka 為例，構(gòu)建 EMQX+Kafka 百萬級吞吐車聯(lián)網(wǎng)平臺：

• 前端車機的連接與消息可通過公有云商提供的負載均衡產(chǎn)品用作域名轉(zhuǎn)發(fā)，如果采用了 TLS/DTLS 的安全認證，可在云上建立四臺 HAProxy/Nginx 服務(wù)器作為證書卸載和負載均衡使用。
• 采用 10 臺 EMQX 組成一個大集群，把一百萬的消息吞吐平均分到每個節(jié)點十萬消息吞吐，同時滿足高可用場景需求。
• 如有離線離線/消息緩存需求，可選用 Redis 作為存儲數(shù)據(jù)庫。
• Kafka 作為總體消息隊列，EMQX 把全量消息通過規(guī)則引擎，轉(zhuǎn)發(fā)給后端 Kafka 集群中。
• 后端 TSP 平臺/OTA 等應(yīng)用通過訂閱 Kafka 的主題接收相應(yīng)的消息，業(yè)務(wù)平臺的控制指令和推送消息可通過 Kafka/API 的方式下發(fā)到 EMQX。

總體架構(gòu)圖

在這一方案架構(gòu)中，EMQX 作為消息中間件具有如下優(yōu)勢，可滿足該場景下的需求：

• 支持千萬級車輛連接、百萬級消息吞吐能力。
• 分布式集群架構(gòu)，穩(wěn)定可靠，支持動態(tài)水平擴展。
• 強大的規(guī)則引擎和數(shù)據(jù)橋接、持久化能力，支持百萬級消息吞吐處理。
• 擁有豐富 API 與認證等系統(tǒng)能順利對接。

百萬吞吐場景驗證

為了驗證上述架構(gòu)的吞吐能力，在條件允許的情況下，我們可以通過以下配置搭建百萬級消息吞吐測試場景。壓測工具可以選用 Benchmark Tools、JMeter 或 XMeter 測試平臺。共模擬 100 萬設(shè)備，每個設(shè)備分別都有自己的主題，每個設(shè)備每秒發(fā)送一次消息，持續(xù)壓測 12 小時。

壓測架構(gòu)圖如下：

性能測試部分結(jié)果呈現(xiàn)：

(1) EMQX 集群 Dashboard 統(tǒng)計

EMQX 規(guī)則引擎中可以看到每個節(jié)點速度為 10 萬/秒的處理速度，10 個節(jié)點總共 100 萬/秒的速度進行。

(2) EMQX 規(guī)則引擎統(tǒng)計

在 Kafka 中可以看到每秒 100 萬的寫入速度，并且一直持續(xù)存儲。

Kafka 管理界面統(tǒng)計

EMQX+InfluxDB 構(gòu)建百萬級吞吐車聯(lián)網(wǎng)平臺

架構(gòu)設(shè)計

采用 EMQX+ 時序數(shù)據(jù)庫的架構(gòu)，同樣可以構(gòu)建百萬級消息吞吐平臺。在本文我們以 InfluxDB 時序數(shù)據(jù)庫為例。

InfluxDB 是一個高性能的時序數(shù)據(jù)庫，被廣泛應(yīng)用于存儲系統(tǒng)的監(jiān)控數(shù)據(jù)、IoT 行業(yè)的實時數(shù)據(jù)等場景。它從時間維度去記錄消息，具備很強寫入和存儲性能，適用于大數(shù)據(jù)和數(shù)據(jù)分析。分析完的數(shù)據(jù)可以提供給后臺應(yīng)用系統(tǒng)進行數(shù)據(jù)支撐。

此架構(gòu)中通過 EMQX 規(guī)則引擎進行消息轉(zhuǎn)發(fā)，InfluxDB 進行消息存儲，對接后端大數(shù)據(jù)和分析平臺，可以更方便地服務(wù)于時序分析。

• 前端設(shè)備的消息通過云上云廠商的負載均衡產(chǎn)品用作域名轉(zhuǎn)發(fā)和負載均衡。
• 本次采用 1 臺 EMQX 作為測試，后續(xù)需要時可以采用多節(jié)點的方式，組成相應(yīng)的集群方案(測試 100 萬可以部署 10 臺 EMQX 集群)。
• 如有離線離線/消息緩存需求，可選用 Redis 作為存儲數(shù)據(jù)庫。
• EMQX 把全量消息通過規(guī)則引擎轉(zhuǎn)發(fā)給后端InfluxDB進行數(shù)據(jù)持久化存儲。
• 后端大數(shù)據(jù)平臺通過 InfluxDB 接收相應(yīng)的消息，對其進行大數(shù)據(jù)分析，分析后再通過 API 的方式把想要的信息傳輸?shù)?EMQX。

總體架構(gòu)圖

場景驗證

如測試架構(gòu)圖中所示，XMeter 壓力機模擬 10 萬 MQTT 客戶端向 EMQX 發(fā)起連接，新增連接速率為每秒 10000，客戶端心跳間隔(Keep Alive)300 秒。所有連接成功后每個客戶端每秒發(fā)送一條 QoS 為 1、Payload 為 200B 的消息，所有消息通過 HTTP InfluxDB 規(guī)則引擎橋過濾篩選并持久化發(fā)至 InfluxDB 數(shù)據(jù)庫。

測試結(jié)果呈現(xiàn)如下：

EMQX Dashboard 統(tǒng)計：

EMQX 規(guī)則引擎統(tǒng)計：

InfluxDB 數(shù)據(jù)庫收到數(shù)據(jù)：

EMQX Dashboard 消息數(shù)統(tǒng)計

單臺 EMQX 服務(wù)器實現(xiàn)了單臺服務(wù)器 10 萬 TPS 的消息吞吐持久化到 InfluxDB 能力。參考 EMQX+Kafka 架構(gòu)的測試場景，將 EMQX 的集群節(jié)點擴展到 10 臺，就可以支持 100 萬的 TPS 消息吞吐能力。

結(jié)語

通過本文，我們介紹了車聯(lián)網(wǎng)場景消息吞吐設(shè)計需要考慮的因素，同時提供了兩種較為主流的百萬級吞吐平臺架構(gòu)設(shè)計方案。面對車聯(lián)網(wǎng)場景下日益增加的數(shù)據(jù)量，希望本文能夠為相關(guān)團隊和開發(fā)者在車聯(lián)網(wǎng)平臺設(shè)計與開發(fā)過程中提供參考。?