個推消息推送基于 Apache Pulsar的優(yōu)先級隊列方案

作者：個推平臺研發(fā)工程師祥子

一.業(yè)務背景

在個推的消息推送場景中，消息隊列在整個系統(tǒng)中占有非常重要的位置?！?/p>

當 APP 有消息推送需求的時候, 會向個推發(fā)送一條消息推送的命令，接到推送需求后，我們會把APP要求推送消息的用戶放入下發(fā)隊列中，進行消息下發(fā);當同時有多個APP進行消息下發(fā)時，難免會出現(xiàn)資源競爭的情況, 因此就產(chǎn)生了優(yōu)先級隊列的需求，在下發(fā)資源固定的情況下, 高優(yōu)先級的用戶需要有更多的下發(fā)資源。

二．基于 Kafka 的優(yōu)先級隊列方案

針對以上場景，個推基于 Kafka 設計了第一版的優(yōu)先級隊列方案。Kafka 是 LinkedIn 開發(fā)的一個高性能、分布式消息系統(tǒng);Kafka 在個推有非常廣泛的應用，如日志收集、在線和離線的消息推送分發(fā)等。

架構(gòu)　　

在該方案中，個推將優(yōu)先級統(tǒng)一設定為高、中、低三個級別。具體操作方案如下：

1. 對某個優(yōu)先級根據(jù) task (單次推送任務) 維度，存入不同的 Topic，一個 task 只寫入一個 Topic，一個 Topic 可存多個 task;

2. 消費模塊根據(jù)優(yōu)先級配額(如 6：3：1)，獲取不同優(yōu)先級的消息推送數(shù)，同一優(yōu)先級輪詢獲取消息;這樣既保證了高優(yōu)先級用戶可以更快地發(fā)送消息，又避免了低優(yōu)先級用戶出現(xiàn)沒有下發(fā)的情況。

Kafka 方案遇到的問題

隨著個推業(yè)務的不斷發(fā)展，接入的 APP 數(shù)量逐漸增多，消息推送的需求也越來越大，第一版的優(yōu)先級方案也逐漸暴露出一些問題：

1. 當相同優(yōu)先級的 APP 在同一時刻消息推送任務越來越多時，后面進入的 task 消息會因為前面 task 消息還存在隊列情況而出現(xiàn)延遲。如下圖所示, 當 task1 消息量過大時，在 task1 消費結(jié)束前，taskN 將一直處于等待狀態(tài)。　　

2. Kafka 在 Topic 數(shù)量由 64 增長到 256 時，吞吐量下降嚴重，Kafka 的每個 Topic、每個分區(qū)都會對應一個物理文件。當 Topic 數(shù)量增加時，消息分散的落盤策略會導致磁盤 IO 競爭激烈，因此我們不能僅通過增加 Topic 數(shù)量來緩解第一點中的問題。

基于上述問題，個推進行了新一輪的技術(shù)選型, 我們需要可以創(chuàng)建大量的Topic, 同時吞吐性能不能比 Kafka 遜色。經(jīng)過一段時間的調(diào)研，Apache Pulsar 引起了我們的關注。

三．為什么是 Pulsar

Apache Pulsar 是一個企業(yè)級的分布式消息系統(tǒng)，最初由 Yahoo 開發(fā)，在 2016 年開源，并于2018年9月畢業(yè)成為 Apache 基金會的頂級項目。Pulsar 已經(jīng)在 Yahoo 的生產(chǎn)環(huán)境使用了三年多，主要服務于Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存儲)。

架構(gòu)　　　　

Topic 數(shù)量

Pulsar 可以支持百萬級別 Topic 數(shù)量的擴展，同時還能一直保持良好的性能。

Topic 的伸縮性取決于它的內(nèi)部組織和存儲方式。Pulsar 的數(shù)據(jù)保存在 bookie(BookKeeper 服務器)上，處于寫狀態(tài)的不同 Topic 的消息，在內(nèi)存中排序，最終聚合保存到大文件中，在 Bookie 中需要更少的文件句柄。另一方面 Bookie 的 IO 更少依賴于文件系統(tǒng)的 Pagecache，Pulsar 也因此能夠支持大量的主題。

消費模型

Pulsar 支持三種消費模型：Exclusive、Shared 和 Failover。　　

* Exclusive (獨享)：一個 Topic 只能被一個消費者消費。Pulsar 默認使用這種模式。

* Shared (共享)：共享模式，多個消費者可以連接到同一個 Topic，消息依次分發(fā)給消費者。當一個消費者宕機或者主動斷開連接時，那么分發(fā)給這個消費者的未確認( ack)的消息會得到重新調(diào)度，分發(fā)給其他消費者。

*Failover (災備)：一個訂閱同時只有一個消費者，可以有多個備份消費者。一旦主消費者故障，則備份消費者接管。不會出現(xiàn)同時有兩個活躍的消費者。

Exclusive和Failover訂閱，僅允許一個消費者來使用和消費每個訂閱的Topic。這兩種模式都按 Topic 分區(qū)順序使用消息。它們最適用于需要嚴格消息順序的流(Stream)用例。

Shared 允許每個主題分區(qū)有多個消費者。同一個訂閱中的每個消費者僅接收Topic分區(qū)的一部分消息。Shared最適用于不需要保證消息推送的順序隊列(Queue)的使用模式，并且可以按照需要任意擴展消費者的數(shù)量。

存儲

Pulsar 引入了 Apache BookKeeper 作為存儲層，BookKeeper 是一個專門為實時系統(tǒng)優(yōu)化過的分布式存儲系統(tǒng)，具有可擴展、高可用、低延遲等特性。具體介紹，請參考[BookKeeper官網(wǎng)](https://github.com/apache/bookkeeper)。

Segment

BookKeeper 以 Segment(在 BookKeeper 內(nèi)部被稱作 ledger)作為存儲的基本單元。從 Segment 到消息粒度，都會均勻分散到 BookKeeper 的集群中。這種機制保證了數(shù)據(jù)和服務均勻分散在BookKeeper集群中。

Pulsar 和 Kafka 都是基于 partition 的邏輯概念來做 Topic存儲的。最根本的不同是，Kafka 的物理存儲是以 partition 為單位的，每個 partition 必須作為一個整體(一個目錄)存儲在某個 broker 上。 而 Pulsar 的 partition 是以 segment 作為物理存儲的單位，每個 partition 會再被打散并均勻分散到多個 bookie 節(jié)點中。

這樣的直接影響是，Kafka 的 partition 的大小，受制于單臺 broker 的存儲;而 Pulsar 的 partition 則可以利用整個集群的存儲容量?！　?/p>

擴容

當 partition 的容量達到上限后，需要擴容的時候，如果現(xiàn)有的單臺機器不能滿足，Kafka 可能需要添加新的存儲節(jié)點，并將 partition 的數(shù)據(jù)在節(jié)點之間搬移達到 rebalance 的狀態(tài)。

而 Pulsar 只需添加新的 Bookie 存儲節(jié)點即可。新加入的節(jié)點由于剩余空間大，會被優(yōu)先使用，接收更多的新數(shù)據(jù);整個擴容過程不涉及任何已有數(shù)據(jù)的拷貝和搬移。

Broker 故障

Pulsar 在單個節(jié)點失敗時也會體現(xiàn)同樣的優(yōu)勢。如果 Pulsar 的某個服務節(jié)點 broker 失效，由于 broker 是無狀態(tài)的，其他的 broker 可以很快接管 Topic，不會涉及 Topic 數(shù)據(jù)的拷貝;如果存儲節(jié)點 Bookie 失效，在集群后臺中，其他的 Bookie 會從多個 Bookie 節(jié)點中并發(fā)讀取數(shù)據(jù)，并對失效節(jié)點的數(shù)據(jù)自動進行恢復，對前端服務不會造成影響?！　?/p>

Bookie 故障

Apache BookKeeper 中的副本修復是 Segment(甚至是 Entry)級別的多對多快速修復。這種方式只會復制必須的數(shù)據(jù)，這比重新復制整個主題分區(qū)要精細。如下圖所示，當錯誤發(fā)生時， Apache BookKeeper 可以從 bookie 3 和 bookie 4 中讀取 Segment 4 中的消息，并在 bookie 1 處修復 Segment 4。所有的副本修復都在后臺進行，對 Broker 和應用透明。

當某個 Bookie 節(jié)點出錯時，BookKeeper 會自動添加可用的新 Bookie 來替換失敗的 Bookie，出錯的 Bookie 中的數(shù)據(jù)在后臺恢復，所有 Broker 的寫入不會被打斷，而且不會犧牲主題分區(qū)的可用性?！　?/p>

四．基于 Pulsar 的優(yōu)先級隊列方案　　

在設計思路上，Pulsar 方案和 Kafka 方案并沒有多大區(qū)別。但在新方案中，個推技術(shù)團隊借助 Pulsar 的特性，解決了 Kafka 方案中存在的問題。

1. 根據(jù) task 動態(tài)生成 Topic，保證了后進入的 task 不會因為其他 task 消息推送的堆積而造成等待情況。

2. 中高優(yōu)先級 task 都獨享一個 Topic，低優(yōu)先級 task 共享 n 個 Topic。

3. 相同優(yōu)先級內(nèi)，各個 task 輪詢讀取消息，配額滿后流轉(zhuǎn)至下一個優(yōu)先級。

4. 相同優(yōu)先級內(nèi), 各個 task 可動態(tài)調(diào)整 quota，在相同機會內(nèi)，可讀取更多消息推送的內(nèi)容。

5. 利用 Shared 模式, 可以動態(tài)添加刪除 consumer，且不會觸發(fā) Rebalance 情況。

6. 利用 BookKeeper 特性，可以更靈活的添加存儲資源。　　

Pulsar 引入了 ApacheBookKeeper 作為存儲層，BookKeeper 是一個專門為實時系統(tǒng)優(yōu)化過的分布式存儲系統(tǒng)，具有可擴展、高可用、低延遲等特性。具體介紹，請參考[BookKeeper官網(wǎng)]

五．Pulsar 其他實踐

1. 不同 subscription 之間相對獨立，如果想要重復消費某個 Topic 的消息，需要使用不同的 subscriptionName 訂閱;但是一直增加新的 subscriptionName，backlog 會不斷累積。

2. 如果 Topic 無人訂閱，發(fā)給它的消息默認會被刪除。因此如果 producer 先發(fā)送，consumer 后接收，一定要確保 producer 發(fā)送之前，Topic 有 subscription 存在(哪怕 subscribe 之后 close 掉)，否則這段時間發(fā)送的消息會導致無人處理。

3. 如果既沒有人發(fā)送消息，又沒有人訂閱消息，一段時間后 Topic 會自動刪除。

4. Pulsar 的 TTL 等設置，是針對整個 namespace 起效的，無法針對單個 Topic。

5. Pulsar 的鍵都建立在 zookeeper 的根目錄上，在初始化時建議增加總節(jié)點名。

6. 目前 Pulsar 的 java api 設計，消息默認需要顯式確認，這一點跟 Kafka 不一樣。

7. Pulsar dashboard 上的 storage size 和 prometheus 上的 storage size (包含副本大小) 概念不一樣。

8. 把 `dbStorage_rocksDB_blockCacheSize` 設置的足夠大;當消息推送的體量大，出現(xiàn) backlog 大量堆積時, 使用默認大小(256M)會出現(xiàn)讀耗時過大情況，導致消費變慢。

9. 使用多 partition，提高吞吐。

10. 在系統(tǒng)出現(xiàn)異常時，主動抓取 stats 和 stats-internal，里面有很多有用數(shù)據(jù)。

11. 如果業(yè)務中會出現(xiàn)單 Topic 體量過大的情況，建議把 `backlogQuotaDefaultLimitGB` 設置的足夠大(默認10G), 避免因為默認使用 `producer_request_hold` 模式出現(xiàn) block producer 的情況;當然可以根據(jù)實際業(yè)務選擇合適的 `backlogQuotaDefaultRetentionPolicy`。

12. 根據(jù)實際業(yè)務場景主動選擇 backlog quota。

13. prometheus 內(nèi)如果發(fā)現(xiàn)讀耗時為空情況，可能是因為直接讀取了緩存數(shù)據(jù);Pulsar 在讀取消息推送內(nèi)容時會先讀取 write cache, 然后讀取 read cache;如果都沒有命中, 則會在 RocksDB 中讀取條目位子后，再從日志文件中讀取該條目。

14. 寫入消息時, Pulsar 會同步寫入 journal 和 write cache;write cache 再異步寫入日志文件和 RocksDB;所以有資源的話，建議 journal 盤使用SSD。

總結(jié)

現(xiàn)在, 個推針對優(yōu)先級中間件的改造方案已經(jīng)在部分現(xiàn)網(wǎng)業(yè)務中試運行，希望通過這樣的模式可以解決消息推送下發(fā)中的一些問題，而對于 Pulsar 的穩(wěn)定性，我們還在持續(xù)關注中。

作為一個 2016 年才開源的項目, Pulsar 擁有非常多吸引人的特性，也彌補了其他競品的短板，例如跨地域復制、多租戶、擴展性、讀寫隔離等。盡管在業(yè)內(nèi)使用尚不廣泛, 但從現(xiàn)有的特性來說, Pulsar 表現(xiàn)出了取代 Kafka 的趨勢。在使用 Pulsar 過程中，我們也遇到了一些問題, 在此特別感謝翟佳和郭斯杰(兩位均為StreamNative的核心工程師、開源項目Apache Pulsar的PMC成員)給我們提供的支持和幫助。

參考文獻：

[1] [比拼 Kafka, 大數(shù)據(jù)分析新秀 Pulsar 到底好在哪]

[2] [開源實時數(shù)據(jù)處理系統(tǒng)Pulsar：一套搞定Kafka+Flink+DB

生成海報

免責聲明：本網(wǎng)站內(nèi)容主要來自原創(chuàng)、合作伙伴供稿和第三方自媒體作者投稿，凡在本網(wǎng)站出現(xiàn)的信息，均僅供參考。本網(wǎng)站將盡力確保所提供信息的準確性及可靠性，但不保證有關資料的準確性及可靠性，讀者在使用前請進一步核實，并對任何自主決定的行為負責。本網(wǎng)站對有關資料所引致的錯誤、不確或遺漏，概不負任何法律責任。任何單位或個人認為本網(wǎng)站中的網(wǎng)頁或鏈接內(nèi)容可能涉嫌侵犯其知識產(chǎn)權(quán)或存在不實內(nèi)容時，應及時向本網(wǎng)站提出書面權(quán)利通知或不實情況說明，并提供身份證明、權(quán)屬證明及詳細侵權(quán)或不實情況證明。本網(wǎng)站在收到上述法律文件后，將會依法盡快聯(lián)系相關文章源頭核實，溝通刪除相關內(nèi)容或斷開相關鏈接。