共計 3048 個字符，預(yù)計需要花費 8 分鐘才能閱讀完成。
如何進行 Apache Pulsar 延遲消息投遞，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。
導(dǎo)語
Apache Pulsar 是一個多租戶、高性能的服務(wù)間消息傳輸解決方案，支持多租戶、低延時、讀寫分離、跨地域復(fù)制、快速擴容、靈活容錯等特性。騰訊數(shù)據(jù)平臺部 MQ 團隊對 Pulsar 做了深入調(diào)研以及大量的性能和穩(wěn)定性方面優(yōu)化，目前已經(jīng)在騰訊云消息隊列 TDMQ 落地上線。本文主要介紹 Pulsar 延遲消息投遞的實現(xiàn)，希望與大家一同交流。
一、什么是延遲消息投遞
延遲消息投遞在 MQ 應(yīng)用場景中十分普遍，它是指消息在發(fā)送到 MQ 服務(wù)端后并不會立馬投遞，而是根據(jù)消息中的屬性延遲固定時間后才投遞給消費者，一般分為定時消息和延遲消息兩種：
定時消息：Producer 將消息發(fā)送到 MQ 服務(wù)端，但并不期望這條消息立馬投遞，而是推遲到在當(dāng)前時間點之后的某一個時間投遞到 Consumer 進行消費。
延遲消息：Producer 將消息發(fā)送到 MQ 服務(wù)端，但并不期望這條消息立馬投遞，而是延遲一定時間后才投遞到 Consumer 進行消費。
目前在業(yè)界，騰訊云的 CMQ 和阿里云的 RocketMQ 也都支持延遲消息投遞：
CMQ：將消息延遲期間定義為”飛行狀態(tài)“，可通過設(shè)置 DelaySeconds 配置延遲范圍，取值范圍為 0 – 3600 秒，即消息最長不可見時長為 1 小時。
RocketMQ：開源版本延遲消息臨時存儲在一個內(nèi)部主題中，支持特定的 level，例如定時 5s，10s，1m 等，商業(yè)版本支持任意時間精度。
開源的 NSQ、RabbitMQ、ActiveMQ 和 Pulsar 也都內(nèi)置了延遲消息的處理能力。雖然每個 MQ 項目的使用和實現(xiàn)方式不同，但核心實現(xiàn)思路都一樣：Producer 將一個延遲消息發(fā)送到某個 Topic 中，Broker 將延遲消息放到臨時存儲進行暫存，延遲跟蹤服務(wù)（Delayed Tracker Service）會檢查消息是否到期，將到期的消息進行投遞。
二、延遲消息投遞的使用場景
延遲消息投遞是要暫緩對當(dāng)前消息的處理，在未來的某個時間點再觸發(fā)投遞，實際的應(yīng)用場景非常多，比如異常檢測重試、訂單超時取消、預(yù)約提醒等。
服務(wù)請求異常，需要將異常請求放到單獨的隊列，隔 5 分鐘后進行重試；
用戶購買商品，但一直處于未支付狀態(tài)，需要定期提醒用戶支付，超時則關(guān)閉訂單；
面試或者會議預(yù)約，在面試或者會議開始前半小時，發(fā)送通知再次提醒；
TDMQ 最近就有個使用 Pulsar 延遲消息的 Case：業(yè)務(wù)要對兩套系統(tǒng)的日志消息進行關(guān)聯(lián)，其中一套系統(tǒng)由于查詢 Hbase 可能會超時或失敗，需要將失敗的關(guān)聯(lián)任務(wù)在集群空閑的時候再次調(diào)度。
三、如何使用 Pulsar 延遲消息投遞
Pulsar 最早是在 2.4.0 引入了延遲消息投遞的特性，在 Pulsar 中使用延遲消息，可以精確指定延遲投遞的時間，有 deliverAfter 和 deliverAt 兩種方式。其中 deliverAt 可以指定具體的時間戳；deliverAfter 可以指定在當(dāng)前多長時間后執(zhí)行。兩種方式的本質(zhì)是一樣的，Client 會計算出時間戳送到 Broker。
1. deliverAfter 發(fā)送
producer.newMessage() .deliverAfter(long time, TimeUnit unit) .send();
2. deliverAt 發(fā)送
producer.newMessage() .deliverAt(long timestamp) .send();在 Pulsar 中，可以支持跨度很大的延時消息，比方說一個月、半年；同時在一個 Topic 里，既支持延時消息，也支持非延時消息。下圖展示了 Pulsar 中延遲消息的具體過程：
producer 發(fā)送的 m1/m3/m4/m5 有不同的延遲時間，m2 是不需要延遲投遞的正常消息，consumer 消費時會根據(jù)不同的延遲時間進行 ack。
四、Pulsar 延遲消息投遞實現(xiàn)原理
從上面的使用方式可以看出，Pulsar 支持的是秒級精度的延遲消息投遞，不同于開源 RocketMQ 支持固定時間 level 的延遲。
Pulsar 實現(xiàn)延遲消息投遞的方式比較簡單，所有延遲投遞的消息會被 Delayed Message Tracker 記錄對應(yīng)的 index。index 是由 timestamp | LedgerID | EntryID 三部分組成，其中 LedgerID | EntryID 用于定位該消息，timestamp 除了記錄需要投遞的時間，還用于 delayed index 優(yōu)先級隊列排序。
Delayed Message Tracker 在堆外內(nèi)存維護著一個 delayed index 優(yōu)先級隊列，根據(jù)延遲時間進行堆排序，延遲時間最短的會放在頭上，時間越長越靠后。consumer 在消費時，會先去 Delayed Message Tracker 檢查，是否有到期需要投遞的消息，如果有到期的消息，則從 Tracker 中拿出對應(yīng)的 index，找到對應(yīng)的消息進行消費；如果沒有到期的消息，則直接消費正常的消息。
如果集群出現(xiàn) Broker 宕機或者 topic 的 ownership 轉(zhuǎn)移，Pulsar 會重建 delayed index 隊列，來保證延遲投遞的消息能夠正常工作。
五、Pulsar 延遲消息投遞面臨的挑戰(zhàn)
從 Pulsar 的延遲消息投遞實現(xiàn)原理可以看出，該方法簡單高效，對 Pulsar 內(nèi)核侵入性較小，可以支持到任意時間的延遲消息。但同時發(fā)現(xiàn)，Pulsar 的實現(xiàn)方案無法支持大規(guī)模使用延遲消息，主要有以下兩個原因：
1. delayed index 隊列受到內(nèi)存限制
一條延遲消息的 delayed index 由三個 long 組成，對于小規(guī)模的延遲消息來說，內(nèi)存開銷并不大。但由于 index 隊列是 subscription 級別，對于 topic 的同一個 partition 來說，有多少個 subscription 就需要維護多少個 index 隊列；同時，由于延遲消息越多、延遲的時間越長，index 隊列內(nèi)存占用也會更多。
2. delayed index 隊列重建時間開銷
上面有提到，如果集群出現(xiàn) Broker 宕機或者 topic 的 ownership 轉(zhuǎn)移，Pulsar 會重建 delayed index 隊列。對于跨度時間長的大規(guī)模延遲消息，重建時間可能會到小時級別。為了減小 delayed index 隊列重建時間，雖然可以給 topic 分更多的 partition 提高重建的并發(fā)度，但沒有徹底解決重建時間開銷問題。
六、Pulsar 延遲消息投遞未來工作
Pulsar 目前的延遲消息投遞方案簡單高效，但處理大規(guī)模延遲消息時仍然存在風(fēng)險。關(guān)于延遲消息投遞，社區(qū)和數(shù)據(jù)平臺部 MQ 團隊下一步將聚焦在支持大規(guī)模延遲消息。目前討論的方案是在 delayed index 隊列加入時間分區(qū)，Broker 只加載當(dāng)前較近的時間片 delayed index 到內(nèi)存，其余時間片分區(qū)持久化磁盤，示例圖如下圖所示：
上圖中，我們按 5 分鐘的間隔對 delayed index 隊列進行分區(qū)，m5 和 m1 放在了 time partition 1，由于延遲時間最近，放在了內(nèi)存；m4 和 m3 在 time partition 2，延遲時間比較靠后，index 存儲在了磁盤。該方案不僅可以減少 delayed index 隊列重建時間開銷，還可以降低對內(nèi)存的依賴。
看完上述內(nèi)容，你們掌握如何進行 Apache Pulsar 延遲消息投遞的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注丸趣 TV 行業(yè)資訊頻道，感謝各位的閱讀！
