2. 程式人生 > >個推基於 Apache Pulsar 的優先級隊列方案

個推基於 Apache Pulsar 的優先級隊列方案

阿新 • • 發佈：2019-04-15
方式 自動刪除 大數據分析 優化 gem 取消 成員 基本 收集

作者：個推平臺研發工程師 祥子

一、業務背景

在個推的推送場景中，消息隊列在整個系統中占有非常重要的位置。
技術分享圖片
當 APP 有推送需求的時候, 會向個推發送一條推送命令，接到推送需求後，我們會把APP要求推送消息的用戶放入下發隊列中，進行消息下發；當同時有多個APP進行消息下發時，難免會出現資源競爭的情況, 因此就產生了優先級隊列的需求，在下發資源固定的情況下, 高優先級的用戶需要有更多的下發資源。

二、基於 Kafka 的優先級隊列方案

針對以上場景，個推基於 Kafka 設計了第一版的優先級隊列方案。Kafka 是 LinkedIn 開發的一個高性能、分布式消息系統；Kafka 在個推有非常廣泛的應用，如日誌收集、在線和離線消息分發等。

架構
技術分享圖片
在該方案中，個推將優先級統一設定為高、中、低三個級別。具體操作方案如下：

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

  2. 消費模塊根據優先級配額(如 6:3:1)，獲取不同優先級的消息數，同一優先級輪詢獲取消息；這樣既保證了高優先級用戶可以更快地發送消息，又避免了低優先級用戶出現沒有下發的情況。

Kafka 方案遇到的問題

隨著個推業務的不斷發展，接入的 APP 數量逐漸增多，第一版的優先級方案也逐漸暴露出一些問題：

  1. 當相同優先級的 APP 在同一時刻推送任務越來越多時，後面進入的 task 消息會因為前面 task 消息還存在隊列情況而出現延遲。如下圖所示, 當 task1 消息量過大時，在task1 消費結束前，taskN 將一直處於等待狀態。
    技術分享圖片
  2. Kafka 在 Topic 數量由 64 增長到 256 時，吞吐量下降嚴重，Kafka 的每個 Topic、每個分區都會對應一個物理文件。當 Topic 數量增加時，消息分散的落盤策略會導致磁盤 IO 競爭激烈，因此我們不能僅通過增加 Topic 數量來緩解第一點中的問題。

基於上述問題，個推進行了新一輪的技術選型, 我們需要可以創建大量的 Topic, 同時吞吐性能不能比 Kafka 遜色。經過一段時間的調研，Apache Pulsar 引起了我們的關註。

三、為什麽是 Pulsar

Apache Pulsar 是一個企業級的分布式消息系統，最初由 Yahoo 開發，在 2016 年開源，並於2018年9月畢業成為 Apache 基金會的頂級項目。Pulsar 已經在 Yahoo 的生產環境使用了三年多，主要服務於Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存儲)。

架構
技術分享圖片

Topic 數量
Pulsar 可以支持百萬級別 Topic 數量的擴展，同時還能一直保持良好的性能。Topic 的伸縮性取決於它的內部組織和存儲方式。Pulsar 的數據保存在 bookie (BookKeeper 服務器)上，處於寫狀態的不同 Topic 的消息，在內存中排序，最終聚合保存到大文件中，在 Bookie 中需要更少的文件句柄。另一方面 Bookie 的 IO 更少依賴於文件系統的 Pagecache，Pulsar 也因此能夠支持大量的主題。

消費模型
Pulsar 支持三種消費模型：Exclusive、Shared 和Failover。
技術分享圖片
Exclusive (獨享)：一個 Topic 只能被一個消費者消費。Pulsar 默認使用這種模式。

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

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

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

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

存儲
Pulsar 引入了 Apache BookKeeper 作為存儲層，BookKeeper 是一個專門為實時系統優化過的分布式存儲系統，具有可擴展、高可用、低延遲等特性。具體介紹，請參考 BookKeeper官網。

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

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

這樣的直接影響是，Kafka 的 partition 的大小，受制於單臺 broker 的存儲；而 Pulsar 的 partition 則可以利用整個集群的存儲容量。
技術分享圖片

擴容
當 partition 的容量達到上限後，需要擴容的時候，如果現有的單臺機器不能滿足，Kafka 可能需要添加新的存儲節點，並將 partition 的數據在節點之間搬移達到 rebalance 的狀態。

而 Pulsar 只需添加新的 Bookie 存儲節點即可。新加入的節點由於剩余空間大，會被優先使用，接收更多的新數據；整個擴容過程不涉及任何已有數據的拷貝和搬移。
技術分享圖片

Broker 故障
Pulsar 在單個節點失敗時也會體現同樣的優勢。如果 Pulsar 的某個服務節點 broker 失效，由於 broker 是無狀態的，其他的 broker 可以很快接管 Topic，不會涉及 Topic 數據的拷貝；如果存儲節點 Bookie 失效，在集群後臺中，其他的 Bookie 會從多個 Bookie 節點中並發讀取數據，並對失效節點的數據自動進行恢復，對前端服務不會造成影響。
技術分享圖片

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

當某個 Bookie 節點出錯時，BookKeeper會自動添加可用的新 Bookie 來替換失敗的 Bookie，出錯的 Bookie 中的數據在後臺恢復，所有 Broker 的寫入不會被打斷，而且不會犧牲主題分區的可用性。
技術分享圖片

四、基於 Pulsar 的優先級隊列方案

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

  1. 根據 task 動態生成 Topic，保證了後進入的 task 不會因為其他 task 消息堆積而造成等待情況。
  2. 中高優先級 task 都獨享一個 Topic，低優先級 task 共享 n 個 Topic。
  3. 相同優先級內，各個 task 輪詢讀取消息，配額滿後流轉至下一個優先級。
  4. 相同優先級內, 各個 task 可動態調整 quota， 在相同機會內，可讀取更多消息。
  5. 利用 Shared 模式, 可以動態添加刪除 consumer，且不會觸發 Rebalance 情況。
  6. 利用 BookKeeper 特性，可以更靈活的添加存儲資源。
    技術分享圖片

五、Pulsar 其他實踐

  1. 不同 subscription 之間相對獨立，如果想要重復消費某個 Topic 的消息，需要使用不同的 subscriptionName 訂閱；但是一直增加新的 subscriptionName，backlog 會不斷累積。
  2. 如果 Topic 無人訂閱，發給它的消息默認會被刪除。因此如果 producer 先發送，consumer 後接收，一定要確保 producer 發送之前，Topic 有 subscription 存在(哪怕 subscribe 之後 close 掉)，否則這段時間發送的消息會導致無人處理。
  3. 如果既沒有人發送消息，又沒有人訂閱消息，一段時間後 Topic 會自動刪除。
  4. Pulsar 的 TTL 等設置，是針對整個 namespace 起效的，無法針對單個 Topic。
  5. Pulsar 的鍵都建立在 zookeeper 的根目錄上，在初始化時建議增加總節點名。
  6. 目前 Pulsar 的 java api 設計，消息默認需要顯式確認，這一點跟 Kafka 不一樣。
  7. Pulsar dashboard 上的 storage size 和 prometheus 上的 storage size (包含副本大小)概念不一樣。
  8. 把dbStorage_rocksDB_blockCacheSize 設置的足夠大；當消息體量大，出現backlog 大量堆積時, 使用默認大小(256M)會出現讀耗時過大情況，導致消費變慢。
  9. 使用多 partition，提高吞吐。
  10. 在系統出現異常時，主動抓取 stats 和 stats-internal，裏面有很多有用數據。
  11. 如果業務中會出現單 Topic 體量過大的情況，建議把 backlogQuotaDefaultLimitGB 設置的足夠大(默認10G), 避免因為默認使用producer_request_hold 模式出現 block producer 的情況；當然可以根據實際業務選擇合適的 backlogQuotaDefaultRetentionPolicy。
  12. 根據實際業務場景主動選擇 backlog quota。
  13. prometheus 內如果發現讀耗時為空情況，可能是因為直接讀取了緩存數據；Pulsar 在讀取消息時會先讀取 write cache, 然後讀取 read cache；如果都沒有命中, 則會在 RocksDB 中讀取條目位子後，再從日誌文件中讀取該條目。
  14. 寫入消息時, Pulsar 會同步寫入 journal 和 write cache；write cache 再異步寫入日誌文件和 RocksDB； 所以有資源的話，建議 journal 盤使用SSD。

六、總結

現在, 個推針對優先級中間件的改造方案已經在部分現網業務中試運行，對於 Pulsar 的穩定性，我們還在持續關註中。
作為一個2016 年才開源的項目，Pulsar 擁有非常多吸引人的特性，也彌補了其他競品的短板，例如跨地域復制、多租戶、擴展性、讀寫隔離等。盡管在業內使用尚不廣泛, 但從現有的特性來說, Pulsar 表現出了取代 Kafka 的趨勢。在使用 Pulsar 過程中，我們也遇到了一些問題, 在此特別感謝翟佳和郭斯傑(兩位均為 Stream Native 的核心工程師、開源項目 Apache Pulsar 的 PMC 成員)給我們提供的支持和幫助。

參考文獻：

[1] 比拼 Kafka, 大數據分析新秀Pulsar 到底好在哪(https://www.infoq.cn/article/1UaxFKWUhUKTY1t_5gPq)

[2] 開源實時數據處理系統Pulsar：一套搞定Kafka+Flink+DB(https://juejin.im/post/5af414365188256717765441)

個推基於 Apache Pulsar 的優先級隊列方案

相關推薦

個推基於 Apache Pulsar 的優先級隊列方案

方式 自動刪除 大數據分析 優化 gem 取消 成員 基本 收集 作者：個推平臺研發工程師 祥子 一、業務背景 在個推的推送場景中，消息隊列在整個系統中占有非常重要的位置。當 APP 有推送需求的時候, 會向個推發送一條推送命令，接到推送需求後，我們會把APP要求推送消息的

個推基於 Apache Pulsar 的優先順序佇列方案

作者：個推平臺研發工程師 祥子 一、業務背景 在個推的推送場景中，訊息佇列在整個系統中佔有非常重要的位置。 當 APP 有推送

python[數據]--隊列，堆，優先級隊列

根據 pop 相同 lambda 隊列 max pytho 優先級 eap 隊列：from collections import deque；實現保存最後幾條歷史記錄，list = deque(maxlen=6),那麽超過六條記錄會刪除之前的記錄。 堆：import hea

優先級隊列與堆排序

缺點 public 概念 對數 col src array 節點 退出 轉自：http://www.cnblogs.com/yangecnu/p/Introduce-Priority-Queue-And-Heap-Sort.html 在很多應用中，我們通常需要按照優先級

java面向對象的棧 隊列 優先級隊列的比較

數據 就會 對象 位置 static 插入 優先級隊列 尾指針 long 　　棧 隊列 有序隊列數據結構的生命周期比那些數據庫類型的結構(比如鏈表，樹)要短得多。在程序操作執行期間他們才被創建，通常用他們去執行某項特殊的任務；當完成任務之後，他們就會被銷毀。這三個數據結構還

二叉堆及優先級隊列

優先級 自底向上 鍵值 由於 ase 數據 二叉 ins err 數據結構既包括各數據存儲的方式和彼此間的關系結構，又含有“添加”、“取出”等對數據的操作，同時也帶有取出和添加數據時的規則。如隊列和棧就是以數據抵達的先後順序來形成這一規則的，但優先級隊列則是以數據內的鍵值作

java數據結構----隊列，優先級隊列

堆數據結構 比較 new n) .com ring ret 插入數據 pan 1.隊列：和棧中的情況不同，隊列中的數據項不總是從數組下標0開始，移除一個數據項後，隊頭指針會指向下標較高的數據項，其特點：先入先出 2.圖解 3.隊列的實現代碼： 　　3.1.Queue

【CCF】交通規劃 Dijstra變形 優先級隊列重載

color space 優先級隊列 ini ios algo OS truct opera 【題意】 給定一個無向圖，求這個圖滿足所有點到頂點的最短路徑不變的最小生成樹 【AC】 註意雙向邊要開2*maxm 註意優先級隊列 #include<iostream>

python隊列queue 之優先級隊列

desc clas prior emp not 定義 div 優先級隊列 col import queue as Q def PriorityQueue_int(): que = Q.PriorityQueue() que.put(10) que

java使用優先級隊列實現哈夫曼編碼

哈夫曼編碼 左右 integer string enc ash 小根堆 rac sta 思路： 構建小根堆 根據小根堆實現哈夫曼樹 根據哈夫曼樹對數據進行編碼 代碼實現如下： /** * @Author: DaleyZou * @Description: 使用jav

隊列 先進先出,先進後出,優先級隊列

sci 正數 元素 指定 元組 print bsp 意思 兩個 import queue #先進先出 q = queue.Queue(3)#設置隊列容量 q.put(1) q.put(a) print(‘查看當前隊列的容量‘,q.qsize) q.put(b) print(

用Python實現數據結構之優先級隊列

sos jsb wim cccccc ont 模塊 uap poj word 優先級隊列 如果我們給每個元素都分配一個數字來標記其優先級，不妨設較小的數字具有較高的優先級，這樣我們就可以在一個集合中訪問優先級最高的元素並對其進行查找和刪除操作了。這樣，我們就引入了

C++優先級隊列表基本算法實現

int ray number while delete pre == stl -- C++優先級隊列表基本算法實現 主要采用鏈式結構，進行數據存儲，然後定義一個最後結點指針數組，將所有優先級最後一個元素的地址保存到這個指針數組中。 #ifndef PriorityQueu

個推基於Docker和Kubernetes的微服務實踐

2016年伊始Docker無比興盛，如今Kubernetes萬人矚目。在這個無比需要創新與速度的時代，由容器、微服務、DevOps構成的雲原生席捲整個IT界。個推針對Web服務場景，基於OpenResty和Node.js搭建了微服務框架，提高了開發效率。在微服務的基礎上，我們結合Doc

個推基於Consul的配置管理

方案 一般來說 動態 互相調用 img https 示例 相關配置 lua實現 作者：個推應用平臺基礎架構高級研發工程師 阿飛在微服務架構體系中，由於微服務眾多，服務之間又有互相調用關系，因此，一個通用的分布式配置管理是必不可少的。一般來說，配置管理需要解決配置集中管理、在

個推基於Zipkin的分散式鏈路追蹤實踐

作者：個推應用平臺基礎架構高階研發工程師 阿飛   01業務背景 隨著微服務架構的流行，系統變得越來越複雜，單體的

Apollo 2.5推基於動態的實時相對地圖解決方案

如何 中心線 lan 精度 poll 視覺 會有 如何工作 adc Apollo 2.5推基於動態的實時相對地圖解決方案 太平洋電腦網 百家號06-2513:19 【PConline資訊】高精地圖對於開發者及測試者來說並不是可以輕易獲得的，制作高精地圖需要大

教你如何使用Java手寫一個基於數組實現的隊列

裏的 void 可用 class 非阻塞 oid 優先級 pub == 一、概述　　隊列，又稱為佇列（queue），是先進先出（FIFO, First-In-First-Out）的線性表。在具體應用中通常用鏈表或者數組來實現。隊列只允許在後端（稱為rear）進行插入操作，在

基於rabbitMQ 消息延時隊列方案 模擬電商超時未支付訂單處理場景

system 方法 只需要 routing new 有效期 conn sta order 前言 傳統處理超時訂單 采取定時任務輪訓數據庫訂單，並且批量處理。其弊端也是顯而易見的；對服務器、數據庫性會有很大的要求，並且當處理大量訂單起來會很力不從心,而且實時性也不是特別好

雙緩沖隊列方案-轉

post .html 數據 wpost 訪問 通過 數量 cpp 生產者-消費者 雙緩沖消息隊列-減少鎖競爭 轉： http://www.cppblog.com/deane/articles/113983.html 在網絡應用服務器端, 為了性能和防止阻塞, 經常會把邏輯處