#本文翻譯自 StreamNative 博客《Achieving Broker Load Balancing with Apache Pulsar》。

作者 Heesung Sohn、王鍇，StreamNative 開發工程師。

校對與整理：StreamNative。

譯者簡介

王中興，eBay 消息中間件團隊成員。

簡介

本文將探討負載均衡在分佈式計算系統中的重要性，並深入分析 Pulsar 處理 Broker 負載均衡的方式。首先我們介紹 Pulsar 中的 Topic-Bundle 分組、Bundle-Broker 歸屬關係以及負載數據模型。然後講解 Pulsar 的負載均衡邏輯，通過時序圖來展示 Bundle 的分配、拆分和縮減過程。通過本文，你將瞭解 Pulsar Broker 是如何做到動態均衡的。

在深入探討 Pulsar Broker 負載均衡的細節之前，我們先簡要討論分佈式計算的挑戰，特別是單體架構系統的挑戰。

分佈式流系統中負載均衡的挑戰

負載均衡是分佈式計算中的一大關鍵挑戰。分佈式系統需要在服務器之間平均分配消息負載，以避免出現服務器過載，從而導致故障並損害集羣性能。一個自然而然的合理選擇是根據主題來對消息進行拆分，因為同一主題（或主題分區）下的消息可以組織到一起並分配給單個邏輯服務器處理。包括 Pulsar 在內的多數分佈式流系統將主題或一組主題視為負載均衡的實體，系統需要在服務器之間平均分配主題或一組主題的消息負載。

當主題負載不可預測時，如何做好主題負載均衡可能會形成挑戰。當某些主題的負載增加時，這些主題必須直接卸載或者重新分區，以便將負載重新分配到其他機器。另一種情況是，某些機器流量非常低，甚至空閒，集羣需要重平衡來避免服務器資源浪費。

動態重平衡在單體架構中可能會很困難，因為消息在同一個有狀態的服務器中處理以及持久化。在單體流式系統中，重平衡通常涉及將消息從一台服務器複製到另一台服務器。管理員必須仔細計算初始主題分佈，儘可能避免將來發生重平衡。在許多情況下，管理員需要仔細編排才能執行主題重平衡操作。

Pulsar 負載均衡概覽

相比之下，Apache Pulsar 則實現了 Broker 的動態負載均衡，無需管理員手工干預。Pulsar 從架構上分離了存儲層和計算層，可以更加靈活地分配 Broker 與主題的映射關係。Pulsar Broker 將消息持久化保存到存儲服務器，當在 Broker 之間重平衡主題時，無需將消息從一個 Broker 複製到另一個 Broker。在這種情況下，新加入的 Broker 只需要查找 Metadata Store 並指向主題消息所在的正確存儲服務器即可。

這裏簡要討論一下 Pulsar 的存儲架構，以便全面地瞭解 Pulsar 的擴展能力。在存儲層，主題消息被分割成多個 Ledger，這些 Ledger 分佈到多個 BookKeeper 服務器，即 Bookie。Pulsar 通過水平擴展 Bookie，即可存儲儘可能多的 Ledger（Segment）條目。

對於高寫入負載，如果所有 Bookie 都已滿，只需增加更多的 Bookie，新的消息條目（即新的 Ledger）即可存儲到這些新的 Bookie 上。通過這種分段設計，在存儲擴展期間 Pulsar 無需從 Bookie 中重新複製舊消息。對於高讀取負載，Pulsar 將消息緩存在 Broker 內存中，所以 Bookie 的讀負載會顯著卸載到 Broker 上，而 Broker 是負載均衡的。你可以在這篇關於對比 Pulsar 與 Kafka ^[1] 的博文中瞭解更多關於 Pulsar 存儲架構以及擴展能力的信息。

在 Bundle 級別分配主題到 Broker

從客户端的角度來看，Pulsar 主題是客户端發佈和消費消息的基本單元。在 Broker 端，單個 Broker 處理所有客户端對某個主題的所有消息請求。主題可以被分區，而分區可以分佈在多個 Broker 上。你可以將主題分區視為一個主題，而將被分區的主題視為一組主題。

由於每個 Broker 只處理一個主題效率較低，所以一般 Broker 需要同時處理多個主題。對於這種多主題歸屬關係，Pulsar 引入了 Bundle 的概念來作為一種中間層組。

在 Pulsar 中，相關的主題可以在邏輯上歸到一個命名空間中。命名空間是一個管理單元，例如可以設置一套配置策略，應用到命名空間中的所有主題上。命名空間內部被分成多個分片，即 Bundle，每個 Bundle 是負載均衡的分配單元。

Pulsar 使用 Bundle 來對主題進行分片，這有助於減少要跟蹤的信息量。例如，Pulsar LoadManager 聚合了主題負載統計信息，比如 Bundle 層的消息速率，這有助於減少要監控的負載樣本數量。此外，Pulsar 需要跟蹤當前是哪個 Broker 服務於特定主題。得益於 Bundle，Pulsar 可以減少維護這種歸屬關係所需的存儲空間。

Pulsar 使用哈希算法將主題映射到 Bundle。如下是一個命名空間包含兩個 Bundle 的示例。

Bundle_Key_Partitions: [0x00000000, 0x80000000, 0xFFFFFFFF]
Bundle1_Key_Range: [0x00000000, 0x80000000)
Bundle2_Key_Range: [0x80000000, 0xFFFFFFFF]

Pulsar 通過 Long hashcode = hash(topicName) 來計算給定主題名的哈希碼。假設 hash(“my-topic”) = 0x0000000F，在已知 Bundle Key 範圍的情況下，Pulsar 可通過 NamespaceBundle getBundle(hashCode) 進行二分搜索，找到主題所屬的 Bundle。在此示例中，“my-topic” 屬於 “Bundle1”。

Bundle 按需動態歸屬到 Broker

Pulsar 計算層（Broker）和存儲層（Bookie）分離的一大優勢是 Pulsar Broker 是無狀態的，基於動態 Bundle 歸屬可以實現良好的水平擴展性。當 Broker 過載後，可以輕鬆地將更多 Broker 加入集羣並重新分配 Bundle 歸屬關係。

Pulsar 使用服務端發現機制來發現給定主題當前的 Bundle-Broker 歸屬關係，將客户端重定向到 Owner Broker 的 URL。這種發現邏輯需要知道：

• • 給定命名空間的 Bundle Key 範圍，以便將主題映射到 Bundle。

• • Bundle-Broker 歸屬關係，以便將客户端定向到當前 Owner；如果尚未分配 Broker，則觸發新一輪歸屬關係分配。

Pulsar 將 Bundle 範圍和歸屬關係存儲在 Metadata Store 中，例如 ZooKeeper 或 etcd，這些信息也會緩存在每個 Broker 中。

負載數據模型

負載均衡決策中至關重要的一點是從 Broker 端收集最新的負載信息。Pulsar 不斷將以下負載數據更新到內存緩存和 Metadata Store 中，並將其複製到 Leader Broker。基於這些負載數據，Leader Broker 執行 Topic-Broker 分配、Bundle 拆分以及卸載邏輯：

• • Bundle 負載數據（Bundle Load Data），包含 Bundle 相關的負載信息，例如 Bundle 的消息輸入/輸出速率。

• • Broker 負載數據（Broker Load Data），包含 Broker 相關的負載信息，例如 CPU、內存以及網絡吞吐量輸入/輸出速率。

負載均衡時序圖

本節將通過時序圖來展示負載均衡邏輯：
1.將主題動態分配到 Broker（閲讀完整文檔^[2]）。
2.拆分過載的 Bundle（閲讀完整文檔^[3]）。
3.從過載的 Broker 中卸載 Bundle（閲讀完整文檔^[4]）。

將主題動態分配到 Broker

假設某客户端想要讀寫主題，現在試圖連接到一個 Broker。該客户端先會連接到一個隨機的 Broker，該 Broker 首先根據主題的哈希碼以及命名空間的 Bundle 範圍搜索匹配的 Bundle。然後，該 Broker 會查詢 Metadata Store，檢查所匹配的 Bundle 是否屬於某 Broker。如果已經歸屬，該 Broker 會將客户端重定向到 Owner URL。否則，會將客户端重定向到 Leader 以進行 Broker 分配。Broker 分配邏輯如下：Leader 首先基於配置好的規則過濾出可用的 Broker 列表，然後隨機選擇一個負載最少的 Broker 分配給 Bundle（如下文第一步所示），並返回該 Broker 的 URL；Leader 將客户端重定向到該 URL，客户端即可連接到分配的 Broker。新的 Broker-Bundle 歸屬關係會在 Metadata Store 中創建一個臨時鎖，一旦 Owner 不可用之後該鎖會自動釋放。

第一步：選定 Broker

這一步從已過濾的可用 Broker 列表中選定一個 Broker，使用 ModularLoadManagerStrategy（默認為 LeastLongTermMessageRate）。LeastLongTermMessageRate 策略計算 Broker 的負載分數，並從分數最小的 Broker 中隨機選擇一個，計分規則如下：

• • 如果 CPU、內存和網絡的最大本地使用率大於 LoadBalancerBrokerOverloadedThresholdPercentage（默認 85%），則設置 score=INF。

• • 否則設置分數為 longTermMsgIn 消息輸入速率加上 longTermMsgOut 消息輸出速率。

拆分過載的 Bundle

Leader Broker 根據 Bundle 負載數據判斷哪些 Bundle 的負載超過閾值（見第二步），並要求 Owner Broker 進行 Bundle 拆分。具體的拆分邏輯如下：Owner Broker 首先計算拆分位置（見第三步），然後據此重新拆分目標 Bundle（見第四步）；完成拆分之後，Owner Broker 將最新的 Bundle 歸屬關係和範圍更新到 Metadata Store 中。如果啟用了 LoadBalancerAutoUnloadSplitBundlesEnabled，新拆分的 Bundle 可以從 Owner Broker 中自動卸載。

第二步：查找目標 Bundle

如果啟用了 loadBalancerAutoBundleSplitEnabled（默認為 true），則啟用自動拆分 Bundle 功能，Leader Broker 會判斷是否有 Bundle 的負載超過 LoadBalancerNamespaceBundle* 配置的閾值。

Defaults
LoadBalancerNamespaceBundleMaxTopics = 1000
LoadBalancerNamespaceBundleMaxSessions = 1000
LoadBalancerNamespaceBundleMaxMsgRate = 30000
LoadBalancerNamespaceBundleMaxBandwidthMbytes = 100
LoadBalancerNamespaceMaximumBundles = 128

如果命名空間中的 Bundle 個數已經達到或超過 MaximumBundles，則會跳過拆分邏輯。

第三步：計算 Bundle 拆分邊界

接下來計算目標 Bundle 的拆分邊界。Bundle 拆分邊界算法可通過 supportedNamespaceBundleSplitAlgorithms 配置。假設某個命名空間有兩個 Bundle 範圍，範圍分佈是 (0x0000, 0X8000, 0xFFFF)，現在要拆分第一個 Bundle 範圍 (0x0000, 0x8000)，可使用如下拆分算法：

RANGE_EQUALLY_DIVIDE_NAME（默認算法）：該算法將目標 Bundle 拆分為具有相同哈希範圍大小的兩個部分，例如要拆分的目標 Bundle 為 (0x0000, 0x8000)，則拆分邊界為 [0x4000]。

TOPIC_COUNT_EQUALLY_DIVIDE：該算法將目標 Bundle 拆分為具有相同主題數的兩個部分。假設在目標 Bundle [0x0000, 0x8000) 中有 6 個主題：

hash(topic1) = 0x0000
hash(topic2) = 0x0005
hash(topic3) = 0x0010
hash(topic4) = 0x0015
hash(topic5) = 0x0020
hash(topic6) = 0x0025

這種情況會在 0x0012 處進行拆分，使左右兩邊的主題數相同。如果要拆分的目標 Bundle 為 [0x0000, 0x8000)，則拆分邊界為 [0x0012]。

第四步：根據邊界拆分 Bundle

示例：
1.給定 Bundle 分區為 [0x0000, 0x8000, 0xFFFF]，拆分邊界為 [0x4000]。
2.拆分後的 Bundle 分佈為 [0x0000, 0x4000, 0x8000, 0xFFFF]。
3.拆分後的 Bundle 範圍為 [[0x0000, 0x4000), [0x4000, 0x8000), [0x8000, 0xFFFF]]。

從過載的 Broker 中縮減（卸載）Bundle

Leader Broker 根據從所有 Broker 中收集的負載信息，識別出哪些 Broker 已經過載，並觸發 Bundle 卸載操作，目的是為了重平衡整個集羣的流量。

Leader Broker 默認使用 ThresholdShedder 策略，計算 CPU、內存以及網絡 IO 之間最大資源使用率的平均值。之後，Leader 找到那些負載高於此基於平均值閾值的 Broker（見第五步）。找到過載的 Broker 之後，Leader 要求它們從高吞吐量的主題開始卸載一些主題 Bundle，直至將 Broker 負載降低到臨界閾值以下。收到卸載請求後，Owner Broker 從 Metadata Store 中移除目標 Bundle 的歸屬信息，並關閉客户端的主題連接。然後客户端重新啟動 Broker 發現機制。最終，Leader 將負載較少的 Broker 分配給被卸載的 Bundle，客户端則連接到新的 Broker。

第五步：ThresholdShedder：查找過載的 Broker

ThresholdShedder 首先使用如下公式計算出所有 Broker 的平均資源使用率。

對每個 Broker： 
    usage =  
    max (
    %cpu * cpuWeight
    %memory * memoryWeight,
    %bandwidthIn * bandwidthInWeight,
    %bandwidthOut * bandwidthOutWeight) / 100;

    usage = x * prevUsage + (1 - x) * usage

    avgUsage = sum(usage) / numBrokers 

如果 Broker 的資源使用率大於 avgUsage + y，則被認為過載。

• • 資源使用率的權重（Weight）默認為 1.0，可通過 loadBalancerResourceWeight 進行配置。

• • 歷史使用率乘子 x 可通過 loadBalancerHistoryResourcePercentage 進行配置。其默認值是 0.9，歷史使用率比最近使用率的權重更大。

• • avgUsage 緩衝值 y 可通過 loadBalancerBrokerThresholdShedderPercentage 進行配置，默認值是 10%。

總結

在本博文中，我們回顧了 Pulsar Broker 的負載均衡邏輯，重點關注其時序圖。我認為 Broker 負載均衡行為有如下幾個要點。

• • Pulsar 通過 Bundle 將主題分組以便於跟蹤，並在 Broker 之間動態地分配和平衡 Bundle。如果特定的 Bundle 發生過載，則自動進行拆分，將分配單元維護在合理的流量水平。

• • Pulsar 將 Broker 全局負載數據（CPU、內存以及網絡使用率）以及 Bundle 負載數據（消息輸入/輸出速率）收集到 Leader Broker，以運行負載均衡算法邏輯：執行 Bundle-Broker 分配、Bundle 拆分和卸載（縮減）。

• • Bundle-Broker 分配邏輯隨機選擇負載最少的 Broker，並將客户端重定向到分配的 Broker URL。Broker-Bundle 歸屬關係會在 Metadata Store 中創建臨時鎖，如果 Owner 不可用（失去歸屬權）則會自動釋放鎖。

• • Bundle 拆分邏輯根據 LoadBalancerNamespaceBundle* 配置的閾值查找目標 Bundle，默認情況下 Bundle 範圍被平均拆分。拆分後，Owner 默認自動卸載新拆分的 Bundle。

• • Bundle 自動卸載邏輯默認使用 LoadSheddingStrategy，根據 CPU、內存以及網絡 IO 的最大資源使用率平均值來查找過載的 Broker。然後 Leader 要求過載的 Broker 卸載一些高負載的主題 Bundle。被卸載的 Bundle 對應的客户端主題連接會被關閉，並重新發起 Bundle-Broker 分配。

引用鏈接

[1] 對比 Pulsar 與 Kafka : http://www.splunk.com/en_us/blog/it/comparing-pulsar-and-kafka-how-a-segment-based-architecture-delivers-better-performance-scalability-and-resilience.html
[2] 閲讀完整文檔: http://pulsar.apache.org/docs/administration-load-balance/#assign-topics-to-brokers-dynamically
[3] 閲讀完整文檔: http://pulsar.apache.org/docs/administration-load-balance/#split-namespace-bundles
[4] 閲讀完整文檔: http://pulsar.apache.org/docs/administration-load-balance/#shed-load-automatically 

---

▼關注「Apache Pulsar」，獲取乾貨與動態 ▼

1. 👇🏻加入 Apache Pulsar 中文交流羣 👇🏻

   ‍‍