負載均衡技術全解析：Pulsar 分布式系統的優秀實踐
背景
Pulsar 有提供一個查詢 Broker 負載的接口：
/**
* Get load for this broker.
*
* @return
* @throws PulsarAdminException
*/
LoadManagerReport getLoadReport() throws PulsarAdminException;
public interface LoadManagerReport extends ServiceLookupData {
ResourceUsage getCpu();
ResourceUsage getMemory();
ResourceUsage getDirectMemory();
ResourceUsage getBandwidthIn();
ResourceUsage getBandwidthOut();
}可以返回一些 broker 的負載數據，比如 CPU、內存、流量之類的數據。
我目前碰到的問題是目前會遇到部分節點的負債不平衡，導致資源占用不均衡，所以想要手動查詢所有節點的負載數據，然后人工進行負載。
理論上這些數據是在運行時實時計算的數據，如果對于單機的倒還好說，每次請求這個接口直接實時計算一次就可以了。
但對于集群的服務來說會有多個節點，目前 Pulsar 提供的這個接口只能查詢指定節點的負載數據，也就是說每次得傳入目標節點的 IP 和端口。
所以我的預期是可以提供一個查詢所有節點負載的接口，已經提了 issue，最近準備寫 Purpose 把這個需求解決了。
實現這個需求的方案有兩種：
- 拿到所有 broker 也就是服務節點信息，依次遍歷調用接口，然后自己組裝信息。
- 從 zookeeper 中獲取負載信息。
理論上第二種更好，第一種實現雖然更簡單，但每次都發起一次 http 請求，多少有些浪費。
第二種方案直接從源頭獲取負載信息，只需要請求一次就可以了。
而正好社區提供了一個命令行工具可以直接打印所有的 broker 負載數據：
pulsar-perf monitor-brokers --connect-string <zookeeper host:port>
分布式系統常用組件
提供的命令行工具其實就是直接從 zookeeper 中查詢的數據。
在分布式系統中需要一個集中的組件來管理各種數據，比如：
- 可以利用該組件來選舉 leader 節點
- 使用該組件來做分布式鎖
- 為分布式系統同步數據
- 統一的存放和讀取某些數據
可以提供該功能的組件其實也不少：
- zookeeper
- etcd
- oxia
Zookeeper 是老牌的分布式協調組件，可以做 leader 選舉、配置中心、分布式鎖、服務注冊與發現等功能。
在許多中間件和系統中都有應用，比如：
- Apache Pulsar 中作為協調中心
- Kafka 中也有類似的作用。
- 在 Dubbo 中作為服務注冊發現組件。
etcd 的功能與 zookeeper 類似，可以用作服務注冊發現，也可以作為 Key Value 鍵值對存儲系統；在 kubernetes 中扮演了巨大作用，經歷了各種考驗，穩定性已經非?？煽苛恕?/p>
Oxia 則是 StreamNative 開發的一個用于替換 Zookeeper 的中間件，功能也與 Zookeeper 類似；目前已經可以在 Pulsar 中替換 Zookeeper，只是還沒有大規模的使用。
Pulsar 中的應用
下面以 Pulsar 為例（使用 zookeeper），看看在這類大型分布式系統中是如何處理負載均衡的。
再開始之前先明確下負載均衡大體上會做哪些事情。
- 首先上報自己節點的負載數據
- Leader 節點需要定時收集所有節點的負載數據。
- CPU、堆內存、堆外內存等通用數據的使用量
- 流出、流入流量
- 一些系統特有的數據，比如在 Pulsar 中就是：
- 每個 broker 中的 topic、consumer、producer、bundle 等數據。
- 這些負載數據中包括：
- 再由 leader 節點讀取到這些數據后選擇負載較高的節點，將數據遷移到負載較低的節點。
以上就是一個完整的負載均衡的流程，下面我們依次看看在 Pulsar 中是如何實現這些邏輯的。
在 Pulsar 中提供了多種負載均衡策略，以下是加載負載均衡器的邏輯：
static LoadManager create(final PulsarService pulsar) {
try {
final ServiceConfiguration conf = pulsar.getConfiguration();
// Assume there is a constructor with one argument of PulsarService.
final Object loadManagerInstance = Reflections.createInstance(conf.getLoadManagerClassName(),
Thread.currentThread().getContextClassLoader());
if (loadManagerInstance instanceof LoadManager) {
final LoadManager casted = (LoadManager) loadManagerInstance;
casted.initialize(pulsar);
return casted;
} else if (loadManagerInstance instanceof ModularLoadManager) {
final LoadManager casted = new ModularLoadManagerWrapper((ModularLoadManager) loadManagerInstance);
casted.initialize(pulsar);
return casted;
}
} catch (Exception e) {
LOG.warn("Error when trying to create load manager: ", e);
}
// If we failed to create a load manager, default to SimpleLoadManagerImpl.
return new SimpleLoadManagerImpl(pulsar);
}默認使用的是 ModularLoadManagerImpl， 如果出現異常那就會使用 SimpleLoadManagerImpl 作為兜底。
他們兩個的區別是 ModularLoadManagerImpl 的功能更全，可以做更為細致的負載策略。
接下來以默認的 ModularLoadManagerImpl 為例講解上述的流程。
上報負載數據
在負載均衡器啟動的時候就會收集節點數據然后進行上報：
public void start() throws PulsarServerException {
try {
String brokerId = pulsar.getBrokerId();
brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + brokerId;
// 收集本地負載數據
updateLocalBrokerData();
// 上報 zookeeper
brokerDataLock = brokersData.acquireLock(brokerZnodePath, localData).join();
} catch (Exception e) {
log.error("Unable to acquire lock for broker: [{}]", brokerZnodePath, e);
throw new PulsarServerException(e);
}
}首先獲取到當前 broker 的 Id 然后拼接一個 zookeeper 節點的路徑，將生成的 localData 上傳到 zookeeper 中。
// 存放 broker 的節點信息
ls /loadbalance/brokers
[broker-1:8080, broker-2:8080]
// 根據節點信息查詢負載數據
get /loadbalance/brokers/broker-1:8080上報的數據：
{"webServiceUrl":"http://broker-1:8080","pulsarServiceUrl":"pulsar://broker-1:6650","persistentTopicsEnabled":true,"nonPersistentTopicsEnabled":true,"cpu":{"usage":7.311714728372232,"limit":800.0},"memory":{"usage":124.0,"limit":2096.0},"directMemory":{"usage":36.0,"limit":256.0},"bandwidthIn":{"usage":0.8324254085661579,"limit":1.0E7},"bandwidthOut":{"usage":0.7155446715644209,"limit":1.0E7},"msgThroughputIn":0.0,"msgThroughputOut":0.0,"msgRateIn":0.0,"msgRateOut":0.0,"lastUpdate":1690979816792,"lastStats":{"my-tenant/my-namespace/0x4ccccccb_0x66666664":{"msgRateIn":0.0,"msgThroughputIn":0.0,"msgRateOut":0.0,"msgThroughputOut":0.0,"consumerCount":2,"producerCount":0,"topics":1,"cacheSize":0}},"numTopics":1,"numBundles":1,"numConsumers":2,"numProducers":0,"bundles":["my-tenant/my-namespace/0x4ccccccb_0x66666664"],"lastBundleGains":[],"lastBundleLosses":[],"brokerVersionString":"3.1.0-SNAPSHOT","protocols":{},"advertisedListeners":{"internal":{"brokerServiceUrl":"pulsar://broker-1:6650"}},"loadManagerClassName":"org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl","startTimestamp":1690940955211,"maxResourceUsage":0.140625,"loadReportType":"LocalBrokerData"}采集數據
public static SystemResourceUsage getSystemResourceUsage(final BrokerHostUsage brokerHostUsage) {
SystemResourceUsage systemResourceUsage = brokerHostUsage.getBrokerHostUsage();
// Override System memory usage and limit with JVM heap usage and limit
double maxHeapMemoryInBytes = Runtime.getRuntime().maxMemory();
double memoryUsageInBytes = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
double memoryUsage = memoryUsageInBytes / MIBI;
double memoryLimit = maxHeapMemoryInBytes / MIBI;
systemResourceUsage.setMemory(new ResourceUsage(memoryUsage, memoryLimit));
// Collect JVM direct memory
systemResourceUsage.setDirectMemory(new ResourceUsage((double) (getJvmDirectMemoryUsed() / MIBI),
(double) (DirectMemoryUtils.jvmMaxDirectMemory() / MIBI)));
return systemResourceUsage;
}會在運行時獲取一些 JVM 和 堆外內存的數據。
收集所有節點數據
作為 leader 節點還需要收集所有節點的負載數據，然后根據一些規則選擇將負載較高的節點移動到負債較低的節點中。
private void updateAllBrokerData() {
// 從 zookeeper 中獲取所有節點
final Set<String> activeBrokers = getAvailableBrokers();
final Map<String, BrokerData> brokerDataMap = loadData.getBrokerData();
for (String broker : activeBrokers) {
try {
String key = String.format("%s/%s", LoadManager.LOADBALANCE_BROKERS_ROOT, broker);
// 依次讀取各個節點的負載數據
Optional<LocalBrokerData> localData = brokersData.readLock(key).get();
if (!localData.isPresent()) {
brokerDataMap.remove(broker);
log.info("[{}] Broker load report is not present", broker);
continue;
}
if (brokerDataMap.containsKey(broker)) {
// Replace previous local broker data.
brokerDataMap.get(broker).setLocalData(localData.get());
} else {
// Initialize BrokerData object for previously unseen
// brokers.
// 將數據寫入到本地緩存
brokerDataMap.put(broker, new BrokerData(localData.get()));
}
} catch (Exception e) {
log.warn("Error reading broker data from cache for broker - [{}], [{}]", broker, e.getMessage());
}
}
// Remove obsolete brokers.
for (final String broker : brokerDataMap.keySet()) {
if (!activeBrokers.contains(broker)) {
brokerDataMap.remove(broker);
}
}
}會從 zookeeper 的節點中獲取到所有的 broker 列表（broker 會在啟動時將自身的信息注冊到 zookeeper 中。）
然后依次讀取各自節點的負載數據，也就是在負載均衡器啟動的時候上報的數據。
篩選出所有 broker 中需要 unload 的 bundle
在 Pulsar 中 topic 是最核心的概念，而為了方便管理大量 topic，提出了一個 Bundle 的概念；Bundle 是一批 topic 的集合，管理 Bundle 自然會比 topic 更佳容易。
所以在 Pulsar 中做負載均衡最主要的就是將負載較高節點中的 bundle 轉移到低負載的 broker 中。
private void updateAllBrokerData() {
final Set<String> activeBrokers = getAvailableBrokers();
final Map<String, BrokerData> brokerDataMap = loadData.getBrokerData();
for (String broker : activeBrokers) {
try {
String key = String.format("%s/%s", LoadManager.LOADBALANCE_BROKERS_ROOT, broker);
Optional<LocalBrokerData> localData = brokersData.readLock(key).get();
if (!localData.isPresent()) {
brokerDataMap.remove(broker);
log.info("[{}] Broker load report is not present", broker);
continue;
}
if (brokerDataMap.containsKey(broker)) {
// Replace previous local broker data.
brokerDataMap.get(broker).setLocalData(localData.get());
} else {
// Initialize BrokerData object for previously unseen
// brokers.
brokerDataMap.put(broker, new BrokerData(localData.get()));
}
} catch (Exception e) {
log.warn("Error reading broker data from cache for broker - [{}], [{}]", broker, e.getMessage());
}
}
// Remove obsolete brokers.
for (final String broker : brokerDataMap.keySet()) {
if (!activeBrokers.contains(broker)) {
brokerDataMap.remove(broker);
}
}
}負載均衡器在啟動的時候就會查詢所有節點的數據，然后寫入到 brokerDataMap 中。
同時也會注冊相關的 zookeeper 事件，當注冊的節點發生變化時（一般是新增或者刪減了 broker 節點）就會更新內存中緩存的負載數據。
之后 leader 節點會定期調用 org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl#doLoadShedding 函數查詢哪些數據需要卸載，然后進行重新負載。
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);最核心的就是調用這個 findBundlesForUnloading 函數，會返回需要卸載 bundle 集合，最終會遍歷這個集合調用 admin API 進行卸載和重平衡。
而這個函數會有多種實現，本質上就是根據傳入的各個節點的負載數據，然后根據自定義的規則返回一批需要卸載的數據。
以默認的 org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder 規則為例：
它是根據帶寬、內存、流量等各個指標的權重算出每個節點的負載值，之后為整個集群計算出一個平均負載值。
以上圖為例：超過 ShedBundles 的數據就需要被卸載掉，然后轉移到低負載的節點中。
所以最左邊節點和超出的 bundle 部分就需要被返回。
具體的計算邏輯如下：
private void filterAndSelectBundle(LoadData loadData, Map<String, Long> recentlyUnloadedBundles, String broker,
LocalBrokerData localData, double minimumThroughputToOffload) {
MutableDouble trafficMarkedToOffload = new MutableDouble(0);
MutableBoolean atLeastOneBundleSelected = new MutableBoolean(false);
loadData.getBundleDataForLoadShedding().entrySet().stream()
.map((e) -> {
String bundle = e.getKey();
BundleData bundleData = e.getValue();
TimeAverageMessageData shortTermData = bundleData.getShortTermData();
double throughput = shortTermData.getMsgThroughputIn() + shortTermData.getMsgThroughputOut();
return Pair.of(bundle, throughput);
}).filter(e ->
!recentlyUnloadedBundles.containsKey(e.getLeft())
).filter(e ->
localData.getBundles().contains(e.getLeft())
).sorted((e1, e2) ->
Double.compare(e2.getRight(), e1.getRight())
).forEach(e -> {
if (trafficMarkedToOffload.doubleValue() < minimumThroughputToOffload
|| atLeastOneBundleSelected.isFalse()) {
selectedBundlesCache.put(broker, e.getLeft());
trafficMarkedToOffload.add(e.getRight());
atLeastOneBundleSelected.setTrue();
}
});
}從代碼里看的出來就是在一個備選集合中根據各種閾值和判斷條件篩選出需要卸載的 bundle。
而 SimpleLoadManagerImpl 的實現如下：
synchronized (currentLoadReports) {
for (Map.Entry<ResourceUnit, LoadReport> entry : currentLoadReports.entrySet()) {
ResourceUnit overloadedRU = entry.getKey();
LoadReport lr = entry.getValue();
// 所有數據做一個簡單的篩選，超過閾值的數據需要被 unload
if (isAboveLoadLevel(lr.getSystemResourceUsage(), overloadThreshold)) {
ResourceType bottleneckResourceType = lr.getBottleneckResourceType();
Map<String, NamespaceBundleStats> bundleStats = lr.getSortedBundleStats(bottleneckResourceType);
if (bundleStats == null) {
log.warn("Null bundle stats for bundle {}", lr.getName());
continue;
}就是很簡單的通過將判斷節點的負載是否超過了閾值 isAboveLoadLevel，然后做一個簡單的排序就返回了。
從這里也看得出來 SimpleLoadManagerImpl 和 ModularLoadManager 的區別，SimpleLoadManagerImpl 更簡單，并沒有提供多個 doLoadShedding 的篩選實現。
總結
總的來說對于無狀態的服務來說，理論上我們只需要做好負載算法即可（輪訓、一致性哈希、低負載優先等）就可以很好的平衡各個節點之間的負載。
而對于有狀態的服務來說，負載均衡就是將負載較高節點中的數據轉移到負載低的節點中。
其中的關鍵就是需要存儲各個節點的負載數據（業界常用的是存儲到 zookeeper 中），然后再由一個 leader 節點從這些節點中根據某種負載算法選擇出負載較高的節點以及負載較低的節點，最終把數據遷移過去即可。
