詳解 Apache Pulsar 消息生命週期

語言: CN / TW / HK
時間 2023-03-20 19:02:30
主題:

文章摘要

本文整理自 Pulsar Summit Asia 2022 騰訊雲高級研發工程師冉小龍的演講《Deep Dive into Apache Pulsar Lifecycle》。Apache Pulsar 中抽象了 Topic 來承載用户發送的消息，一條消息發送到 Topic 中之後會經過 Broker 的計算存儲到 Bookie 中。本文將詳細闡述消息是如何發送到 Broker 並經過 Broker 的計算以及元數據處理最終存儲到 Bookie 中，然後會進一步闡述 Bookie 如何利用垃圾回收機制回收 Topic 中的數據，以及 Broker 中的 TTL 和 Retention 策略如何作用到 Bookie Client 來觸發垃圾回收的機制。

作者簡介

冉小龍，騰訊雲高級研發工程師，Apache Pulsar Committer，RoP maintainer，Apache Pulsar Go Client、Pulsarctl 與 Go Functions 作者與主要維護者。

導讀

本文分為以下幾個部分：

  1. 1. 從用户的視角看消息收發流程
  2. 2. TTL 與 Retention 策略（與消息生命週期息息相關）
  3. 3. 從 Topic 的角度看消息存儲模型
  4. 4. Bookie GC 回收機制
  5. 5. 髒數據如孤兒 Ledger 的產生
  6. 6. 如何清理髒數據
  7. 1、2、3 主要在 Broker 層面分析原理，5 和 6 根據生產環境中遇到的問題來分析髒數據的產生與清理。

用户視角下的消息收發流程

在用户視角下，MQ 可以理解為 Pub-Sub 模型，在 Broker 抽象一個 Topic，消息經由生產者發送到 Topic 中然後進入消費者進行消費。

圖片

首先需要了解兩個概念，Pending Queue 和 Receive Queue。

  • Pending Queue：發送過程中的概念。消息發送時並不是每次直接投遞給 Broker，而是在本地抽象 Pending Queue，所有數據先進入 Pending Queue 再被髮送到 Broker。

  • Receive Queue：接收過程中的概念。同 Pending Queue 原理相同，消息接收時並不是每次直接從 Broker 要數據，而是在本地抽象 Receive Queue，數據按批次進入 Receive Queue，再結合 Pulsar 消息推拉機制不斷地填充 Receive Queue 來調動整體流程。

在 Pulsar 中，Broker 不解析批消息，因此 Broker 無法知道消息是否是批消息，這裏抽象了一個 Entry 的概念，Entry 內可能包含批消息或者非批消息。

下圖是用户視角下更深入的架構圖。生產者和消費者可以理解為 Client 模型，Client 把消息發送給 Broker。Broker 可以理解為 BookKeeper Client，BookKeeper Client 通過增刪改查的操作將數據傳遞給 Bookie。BookKeeper 和 Broker 都有元數據管理中心，目前使用較多的是 ZooKeeper，其內包含所有節點信息，如節點調度信息。

圖片

下面解析一下數據從 Client 到 Broker 再到 BookKeeper 是怎樣的整體流程。首先，BookKeeper 存儲層功能比較單一且純粹。作為一個分佈式日誌文件系統，它暴露給上層系統的、能夠供上層系統調用的僅僅是增刪改查的操作，伴隨這些操作可以觀察從 Client 到 BookKeeper 的操作鏈路：

  • Send -> Broker -> add Entry -> Bookie：發送 Send 命令到 Broker，Broker 向 BookKeeper addEntry
  • Receive -> Broker -> read Entry -> Bookie：發送 Receive 命令到 Broker，Broker 調用 BookKeeper readEntry 接口從 Bookie 中讀取消息
  • Ack -> Broker (TTL) -> move cursor (markDeletePosition) -> Bookie：發送 Ack 命令到 Broker，Broker 會執行 move cursor 操作。Broker 抽象的 Topic 裏面有一條條的消息，Ack 相當於操作 cursor 的行為，指針隨着 Ack 行為移動，此處抽象了 markDeletePosition 的指針。在 markDeletePosition 之前，所有的消息都已被正確消費。
  • Retention -> delete Entry -> Bookie：接收到 Retention 策略後，Broker 觸發 Retention 閾值後會調用 Bookie delete Entry 接口，來刪除 BookKeeper 中數據。delete Entry 是本文重點討論的話題，後文將具體介紹觸發 Retention 策略後，Entry 如何被從 BookKeeper 中刪除。

TTL 與 Retention 策略

首先需要明確 TTL 策略和 Retention 策略的概念。

TTL 策略

TTL 策略指消息在指定時間內沒有被用户 Ack 時會在 Broker 主動 Ack 掉。

Client 在消費者側暴露兩個接口 Receive 和 Ack。當用户消費者接收到消息時，Broker 並不知道此時用户已經正確接收到消息，需要用户手動調用 Ack 告訴 Broker 自己成功接收到了當前消息，所以 Client 要發起 Oneway 的 Ack 請求通知 Broker 進行下一步處理。不論消息是否被推送到 Broker，生產者發送到 Topic 的消息都會產生 TTL（生命週期）。所有消息都在 TTL 內受管控，超出這個時間後 Broker 會代替用户把消息 Ack 掉。

此處需注意，在上述過程中沒有任何與刪除相關的操作，因為 TTL 不涉及與刪除相關的操作。TTL 的作用僅僅是用於 Ack 掉在 TTL 範圍內應被 Ack 的消息，真正刪除的操作與 Pulsar 中抽象出來的 Retention 策略相關。

Retention 策略

Retention 策略指消息被 Ack 之後（消費者 Ack 或者 TTL Ack）繼續在 Bookie 側保留的時間，以 Ledger 為最小操作單元。

消息被 Ack 之後（消費者 Ack 或者 TTL Ack）就歸屬於 Retention 策略，即在 BookKeeper 保留一定時間，比如在離線消息場景下會將數據保留一段時間來進行回查等操作。Retention 以 Ledger 為最小操作單元，刪除即是刪除整個 Ledger。

下面是在 TTL 內 Ack 消息的示意圖。在 T1 時間段有 10 條消息，m1 - m5 是被 Ack 的消息，m6 - m10 是未被 Ack 的消息。在 T2 時間段，假設到達 TTL 的 3 分鐘閾值後消息還沒有被 Ack，m6 - m8 就會被 TTL 策略檢查到，Broker 主動將其 Ack。在 T3 時間段，m6 - m8 已被 Broker Ack。這就是 TTL 策略操作行為與作用範圍。

圖片

Pulsar 內的所有策略都在 Broker 抽象了線程池，週期性地執行線程，比如 TTL 策略或者 Retention 策略默認 5 分鐘檢查一輪。TTL 策略就是根據設置的時間，定期檢查，不斷更新 Cursor 的位置（等價於 Consumer 側暴露的 Ack 接口），將消息過期掉；Retention 策略是檢查 Ledger 的創建時間以及 Entry 的大小來決定是否要刪除某一個 Ledger。

TTL 策略和 Retention 策略的生命週期在時限上有如下規則：

  • TTL 時間 < Retention 時間，消息的生命週期等於 TTL 時間 + Retention 時間。

  • TTL 時間 ≥ Retention 時間，消息的生命週期等於 TTL 時間。在 TTL 檢查時，有一個判斷標準是 Ledger 是否進行切換，如發生切換且達到 TTL 時間，Ledger 會進入 Retention 策略刪除動作。所以如果 TTL 時間 ≥ Retention 時間，消息生命週期就是 TTL 時間。

從 Topic 的角度看消息存儲模型

圖片

講到消息存儲模型，首先接觸到的是 Topic，生產者向這個 Topic 發送消息、消費者從 Topic 消費消息。Topic 內部抽象了 Partition 的概念，一個 Topic 內可以創建多個 Partition，作用是增加併發處理的能力，即一個 Topic 中的消息可以分發到多個 Partition，由多個 Partition 承載 Topic 的服務。

在 Bookie 存儲層，一個 Partition 由多個 Ledger 構成。如圖，Partition 3 下面有 5 個 Ledger。Ledger 裏面存儲的是多條 Entry。如前文所説的 Entry 概念，根據消息是否是批消息，Entry 就可以分為批和非批兩種。如果消息是批消息，那麼 Entry 裏面有多條 Message；如果消息是非批的，那麼一條 Entry 等於一條 Message。這就是 Topic 視角下的存儲模型。

Bookie GC 回收機制

前面三個部分都圍繞 Broker 層，Broker 作為計算層，本質是 Bookie Client，調用 Bookie 側暴露的增刪查的接口來進行相關的操作，操作邏輯簡單。下面將重點介紹 BookKeeper 層如何將數據進行壓縮和回收。

Bookie 壓縮類型

壓縮類型分為兩種：

  • 自動壓縮：Bookie 有周期性執行的 GC Compaction 線程，GC 分為 Minor GC 和 Major GC，後文會詳細介紹兩種 GC 的區別。

  • 手動壓縮：通過 BookKeeper 暴露的 Http 調用 Admin Rest API 接口來觸發 GC 請求。這個操作在日常急救運維中很常見，比如 Bookie 磁盤內存突然大幅度上漲，用户想要緊急回收數據，那麼就可以跳過 Minor GC 和 Major GC 檢查週期，手動觸發 GC 來釋放磁盤空間。

Bookie 壓縮方式

Bookie 的壓縮方式分為兩種：

  • 按照 Entry 大小

    • compactionRateByEntries
    • isThrottleByBytes

  • 按照 Entry 數量（默認）

    • compactionRateByEntries

  • 生產環境中推薦按照 Entry 大小壓縮，從實際生產環境的經驗來看，每次壓縮 100MB，曲線相對平穩。為什麼不推薦按照 Entry 數量壓縮呢？首先如前文提到的 Entry 的概念，一個 Entry 可能是單條消息，也可能是批消息（包含很多 Message），因此如果按照數量壓縮的話，每次壓縮的 Message 數量是不一定的。另外，每一個 Message 的 Payload 不同，消息大小不一致會導致每次壓縮大小不同，GC 壓縮回收的曲線不平穩。Bookie GC 佔用磁盤 IO，每一台機器的磁盤 IO 恆定，極端情況下，不平穩的壓縮會映射到 Bookie 主鏈路讀寫流程，影響穩定性。按照 Entry 大小壓縮，壓縮曲線平穩，對穩定性影響較小。

Minor GC 和 Major GC

從代碼實現邏輯上來看，Minor GC 和 Major GC 完全相同，二者區別在於觸發時機和觸發閾值。

Minor GC Major GC
壓縮時間 1h 24h
壓縮閾值比例 20% （minorCompactionThreshold） 80% （majorCompactionThreshold）
GC 執行最大耗時 minorCompactionMaxTimeMillis majorCompactionMaxTimeMillis
  • Minor GC 壓縮時間是 1h，Major GC 壓縮時間是 24h。
  • 壓縮閾值比例的含義是 Bookie 裏面有用數據的佔比。在 Minor GC 內，Bookie 有用數據佔比為 20%；在 Major GC 內，Bookie 有用數據佔比為 80%。當有用數據佔比超過 20% 和 80% 時，不對數據進行回收。Entrylog 裏文件大小固定為 1.1 GB，假設 Major GC 有用數據超過 80%，那麼可以理解為大部分數據都是有用的且不可被刪除，Entrylog 全部保留。剩下的 20% 數據沒必要耗費磁盤 IO 進行回收，通過多佔用一定空間的方式降低磁盤 IO 的損耗。
  • 為了避免一次 GC 執行時間過長，因此設定了 GC 執行最大耗時。超過規定的耗時就會強行中止 GC。

注意：

  • 壓縮閾值比例不可以超過 100%。
  • Minor GC 的閾值必須小於 Major GC。
  • 壓縮時，必須要保證磁盤還有一定的可用空間。

Bookie 壓縮

Bookie 壓縮時，首先需要了解以下幾個概念。（生產環境中配置 DBLedgerStorage，社區目前使用居多。後文所有 GC 回收流程和 BookKeeper 相關內容都在默認此配置的前提下展開。）

  • Metadata Store：元數據存儲中心默認使用 ZooKeeper。我使用的是社區提供的工具 ZK-Web，可以看到 Ledger 路徑下存儲了很多 Ledger。

    圖片

  • LedgerIndex：RocksDB 中存儲的 Ledger 集合。使用 DBLedgerStorage 即相當於用 RocksDB 做 Entrylog 的索引存儲，讀取數據時先讀取 RocksDB 來找到索引數據，然後去 Entrylog 讀 Value。這是一個拿 Key 取 “V” 的操作。

    圖片

  • LedgersMap：當前的單個 EntryLog 中存儲的 Ledger 集合。

  • EntryLogMetaMap：當前 Bookie 下所有 EntryLog，Key 是 Entrylog ID，Value 是 Entrylog Metadata。EntryLogMetaMap 是 EntryLogMeta 的集合，EntryLogMetaMap 中包含 LedgersMap 集合。

有了上面的抽象後，我們就可以進行判斷。EntryLogMetaMap 的 Key 是 Entrylog ID，映射到 LedgersMap 集合。

圖片

在整個壓縮過程中，有三個核心的處理邏輯與函數：

  1. doGcLedgers()：處理 LedgerIndex 的集合（RocksDB），通過集合判斷數據是否可以刪除。

  2. doGcEntryLogs()：處理 LedgersMap 和 EntryLogMetaMap 的集合，以 doGcLedgers() 得出的集合為基準來判斷當前 LedgersMap 中哪些 Ledger 可以刪除，以及當前 EntryLogMetaMap 中哪些 Entrylog 可以刪除。

  3. doCompactionEntryLogs()：在進行完上面兩個步驟後就可以進行具體的刪除操作。 doCompactionEntryLogs() 處理 EntryLog 文件本身是否可以被刪除，對於一個 Key Value 庫來説如何進行刪除也是一門學問。刪除操作不能直接從 Key Value 集合刪除，這樣會造成很多消息空洞（消息不連續）。BookKeeper 中刪除操作是從舊的 EntryLog 文件讀取不可刪除的數據寫入到新的 EntryLog 文件中，相當於在新的 EntryLog 文件中進行備份，因此舊的 EntryLog 文件可以一次性刪除。

前文多次提到了 EntryLog，下面將介紹 BookKeeper 中 EntryLog 如何存儲、存儲了什麼。Entrylog 的構成從上至下核心數據分為三部分。下圖可以幫助大家瞭解 Entrylog 的大致結構，如需精確瞭解，可以閲讀相關源碼。

  • Header：包含指紋信息（BKLO，標識 Entrylog 文件，用於校驗）、BookKeeper 版本、Ledgers Map Offset（Offset 偏移量、如何讀取等）與 Ledgers Count（一個 Entrylog 內 Ledger 的數量）。

  • LedgerEntry List：LedgerEntry 對象，包含 Entry Size、Ledger ID、Entry ID 和 Count。

  • Ledgers Map：包含 Ledgers Map Size、Ledgers Count 和 Ledgers Map Entries。每一個 Ledgers Map Entries 是 Key Value 結構，由 Ledger 映射到 Size。

    圖片

數據回收全流程

有了上面介紹的基礎概念，我們就可以把數據從 Broker 到 BookKeeper 的回收流程串聯起來。

圖片

首先 Client 觸發流程。創建 Topic 建議設置 Retention 策略，不設置的話默認策略是消費完成即刪除該消息。設置 Retention 策略後，Broker 有定期檢查的線程，週期性針對 Topic 執行 Retention 策略。到期可刪除的 Ledger 調用暴露的 Delete Ledger 接口，如圖 Ledger 0 可刪除，即調用 Delete Ledger 刪除 Ledger 0。刪除 Ledger 0 後 ZooKeeper 中移除 Ledger 0 的 ZooKeeper 路徑。這就是完整的刪除流程，上圖不包含返回邏輯。

Delete Ledger 從調用到返回成功的過程中沒有使用 BookKeeper 磁盤上的數據。用户可能會困惑調用接口刪除 Ledger 為何沒有釋放磁盤空間，原因在此，因為刪除操作和 BookKeeper 回收磁盤的操作是完全異步化的。BookKeeper 回收磁盤的操作由 GC Compaction 線程固定進行處理。

那麼，GC Compaction 週期性執行線程如何運行？GC Compaction 週期性執行線程就是 Minor GC 和 Major GC。在操作流程上，首先會獲取 ZooKeeper 內所有 Ledger 列表。因為創建 Ledger 需要向 ZooKeeper 註冊對應的 ZooKeeper 路徑，刪除 Ledger 也需要從 ZooKeeper 上刪除路徑。ZooKeeper 上的 Ledger 路徑最全面也最準確，因此以 Metadata Store (zk) 為基準來獲取所有 Ledger 列表的集合。然後進行 doGcLedgers() 操作，把 RocksDB 中所有 Ledger 列表集合與 ZooKeeper 上獲取的 Ledger 列表集合做比較，找出可以刪除的 Ledger。刪除後進行 doGcEntryLogs() 操作，處理 LedgersMap 和 EntryLogMetaMap 的集合，判斷 EntryLog 中哪些 Ledger 可以刪除。進一步刪除後進行 doCompactionEntryLogs() 操作，最理想的情況下，Entrylog 裏面所有的 Ledger 都可以被刪除，那麼就可以直接清除這個 Entrylog。大部分情況是 Entrylog 裏部分數據可刪、另一部分不可刪，那麼如何判斷是否保留 Entrylog 呢？由 Minor GC 和 Major GC 的壓縮閾值比例決定。

我們結合下圖瞭解如何通過 doGcEntryLogs() 來 doCompactionEntryLogs()。假設 doCompactionEntryLogs() 時通過 Major GC 的閾值判定一部分未達標的數據可以進行回收，那麼 GC Compaction 線程首先從舊的 Entrylog 中檢查 Ledger 是否可以刪除。假定 Ledger 0 和 Ledger 2 可以刪除，Ledger 1 和 Ledger 3 不可以刪除，檢查到可用性佔比後根據閾值判斷 Entrylog 可以刪除，那麼就把 Ledger 1 和 Ledger 3 的有用數據寫入新的 Entrylog 文件，有用數據有備份後就可以刪除舊的 Entrylog 文件。

圖片

此處需要補充一點，創建新的 Entrylog 文件時還有一個動作叫做 Flush。舊的 Entrylog 文件在創建時會產生索引信息，Bookie 裏 Entrylog 在讀取 Entry 時，比如讀取 Entry 0、Ledger 1 的數據，會根據索引信息來追溯對應的 Entrylog。在刪除舊的 Entrylog 文件並創建新的 Entrylog 文件操作完成之後，新的 Entrylog 文件索引信息需要更新到 RocksDB，通知上層的讀請求去尋找新的 Entrylog 文件中生成的十六進制的 ID 來讀取 Entry 0、Ledger 1 的數據。

以上是消息完整的生命週期，包含從 TTL 與 Retention 策略到 Bookie GC 回收機制的全流程。

髒數據的產生

下面介紹在實際生產中遇到的問題。在下圖中，我們監控了每個 Bookie 上的 Entrylog 文件發現，假設設置的 Retention 策略週期為 1 天或 5 天，但是這些 Entrylog 文件已經存在超出 200 天還沒有被刪除。這是異常情況，文件不刪除會一直佔用磁盤空間。經過分析，以下三個情況可能導致髒數據的產生：

圖片

  • Ledger 刪除邏輯出錯，導致孤兒 Ledger 產生：回顧數據回收全流程，Ledger 刪除操作分為兩個部分：從 ZooKeeper 中清理路徑和 GC Compaction 線程清理 Entrylog。社區發起了 PIP[1] 進行雙階段刪除，來保證刪除過程中不會產生孤兒 Ledger。
  • Broker 不會加載不活躍的 Topic，導致 Retention 策略沒有生效：目前社區正在改進該邏輯。BookKeeper 唯一暴露的 Delete Ledger 操作只有在設置 Retention 策略後才能掉入行為。因此如果 Retention 策略沒有生效，Broker 不活躍 Topic 產生的 Ledger 就無法被刪除。
  • GC 回收閾值設置不合理，導致一部分數據無法從 EntryLog 移除：這是上圖中產生存在 200 多天的 Entrylog 的主要原因。根據對用户數據的調配發現，系統沒有按照 80% 的有用數據佔比來設置回收閾值，而是調整為 50%，導致一半的數據一直存在於 Entrylog 中，無法刪除 Entrylog。
  • 存在不活躍的 Cursor（不活躍即是 Sub 下沒有對應的消費者），這些 Cursor 對應的 Ledger 無法被刪除：目前提出的方案是增加校驗邏輯，如果 Cursor 一段時間內不更新則刪除，此方案還有待商榷與驗證。無論以上哪一種情況，都會導致 Ledger 髒數據無法刪除。因此下面我們展開講解如何刪除髒數據。在瞭解刪除髒數據前，需要了解一個概念叫 Custom Metadata。在 Broker 生成或者創建 Ledger 時，可以給 Ledger 設置一部分元數據，即自定義 Ledger 的元數據屬性。下圖是 Pulsar 默認提供的 Custom Metadata，通過 BookKeeper Admin ctl 獲取到的 Pulsar Managed Ledger Base64 信息。這一串屬性反寫出來就是一個 Topic 的信息，只有擁有 Topic 信息才能進行後面的操作。

圖片

通過 Ledger Metadata 可以獲取 Topic 信息，即 Ledger 的 Owner Topic。然後我們就可以開始清除這些髒數據。

清除孤兒 Ledger

清除孤兒 Ledger 使用 Clear Tool 清除工具。過程如下：

  • 從 ZooKeeper Snapshot 中獲取所有的 Ledger 列表（如果線上環境壓力不大，也可以直接連接 ZooKeeper 讀取，不需要使用 Snapshot。）從 ZooKeeper Snapshot 中獲取所有的 Ledger 列表後，通過 BookKeeper Admin 工具獲取 Ledger 的 Custom Metadata。

  • 通過 Custom Metadata 找到該 Ledger 的 Owner Topic，並在 Broker 內查看是否存在該 Topic。

    • 如果 Broker 內 Topic 不存在，Client 首先訪問 Broker 就無法成功。BookKeeper 存儲數據沒有意義，可以直接刪除。

    • 如果 Broker 內 Topic 存在， 就會進一步檢查 Ledger 是否存在，Topic Stats Internal 列表展示了 Topic 內所有 Ledger 的情況，用來確認該 Ledger 是否包含在該 Topic 中。注意，Topic Stats Internal 命令有時候可以可以獲取到 Ledger 列表，有時無法獲取，解決方法是重複獲取，如果仍獲取不到，那麼將判定為列表不存在。

      ![圖片](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/889dc444f8c2467bb8f07aa5f0accd29~tplv-k3u1fbpfcp-zoom-1.image "null")

Topic 所有的屬性以及 Topic Stats Internal 等指標信息都是 Broker 向 ZooKeeper 獲取的。以上檢查都過後就可以從 BookKeeper 中刪除 Ledger。Ledger 刪除邏輯和前文回收流程相同，首先刪除 Ledger 的 ZooKeeper 路徑，Ledger 佔用的磁盤空間通過 GC Compaction 線程走異步流程進行刪除。

此外，Schema 和 Cursor 信息也會使用 Ledger 來存儲。下圖中有一個信息是 Pulsar Schema ID，如果用户指定了 Schema 是 String、Json，那麼就會產生也對應 Ledger 的 Schema 屬性，ZooKeeper 下面也會存儲 Schema 信息。檢查 Stats Internal 時可以獲取到 Schema Ledger 和 Cursor Ledger，需要仔細查看。

圖片

注意：清理髒數據時一定要備份。ZooKeeper Snapshot 備份可以在錯誤刪除後恢復數據。

總結

文章從用户視角出發，講述了消息存儲到 Bookie 中的流程，並闡述 Bookie 的垃圾回收機制，以及 TTL 和 Retention 策略如何作用到 Bookie Client 觸發垃圾回收機制。希望可以為用户在生產環境中的操作提供參考。

引用鏈接

[1] PIP: https://github.com/apache/pulsar/issues/16569

「」