負載均衡技術(shù)全解析：Pulsar 分布式系統(tǒng)的優(yōu)秀實踐
背景
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 的負載數(shù)據(jù)，比如 CPU、內(nèi)存、流量之類的數(shù)據(jù)。
我目前碰到的問題是目前會遇到部分節(jié)點的負債不平衡，導(dǎo)致資源占用不均衡，所以想要手動查詢所有節(jié)點的負載數(shù)據(jù)，然后人工進行負載。
理論上這些數(shù)據(jù)是在運行時實時計算的數(shù)據(jù)，如果對于單機的倒還好說，每次請求這個接口直接實時計算一次就可以了。
但對于集群的服務(wù)來說會有多個節(jié)點，目前 Pulsar 提供的這個接口只能查詢指定節(jié)點的負載數(shù)據(jù)，也就是說每次得傳入目標節(jié)點的 IP 和端口。
所以我的預(yù)期是可以提供一個查詢所有節(jié)點負載的接口，已經(jīng)提了 issue，最近準備寫 Purpose 把這個需求解決了。
實現(xiàn)這個需求的方案有兩種：
- 拿到所有 broker 也就是服務(wù)節(jié)點信息，依次遍歷調(diào)用接口，然后自己組裝信息。
- 從 zookeeper 中獲取負載信息。
理論上第二種更好，第一種實現(xiàn)雖然更簡單，但每次都發(fā)起一次 http 請求，多少有些浪費。
第二種方案直接從源頭獲取負載信息，只需要請求一次就可以了。
而正好社區(qū)提供了一個命令行工具可以直接打印所有的 broker 負載數(shù)據(jù)：
pulsar-perf monitor-brokers --connect-string <zookeeper host:port>
分布式系統(tǒng)常用組件
提供的命令行工具其實就是直接從 zookeeper 中查詢的數(shù)據(jù)。
在分布式系統(tǒng)中需要一個集中的組件來管理各種數(shù)據(jù)，比如：
- 可以利用該組件來選舉 leader 節(jié)點
- 使用該組件來做分布式鎖
- 為分布式系統(tǒng)同步數(shù)據(jù)
- 統(tǒng)一的存放和讀取某些數(shù)據(jù)
可以提供該功能的組件其實也不少：
- zookeeper
- etcd
- oxia
Zookeeper 是老牌的分布式協(xié)調(diào)組件，可以做 leader 選舉、配置中心、分布式鎖、服務(wù)注冊與發(fā)現(xiàn)等功能。
在許多中間件和系統(tǒng)中都有應(yīng)用，比如：
- Apache Pulsar 中作為協(xié)調(diào)中心
- Kafka 中也有類似的作用。
- 在 Dubbo 中作為服務(wù)注冊發(fā)現(xiàn)組件。
etcd 的功能與 zookeeper 類似，可以用作服務(wù)注冊發(fā)現(xiàn)，也可以作為 Key Value 鍵值對存儲系統(tǒng)；在 kubernetes 中扮演了巨大作用，經(jīng)歷了各種考驗，穩(wěn)定性已經(jīng)非?？煽苛恕?/p>
Oxia 則是 StreamNative 開發(fā)的一個用于替換 Zookeeper 的中間件，功能也與 Zookeeper 類似；目前已經(jīng)可以在 Pulsar 中替換 Zookeeper，只是還沒有大規(guī)模的使用。
Pulsar 中的應(yīng)用
下面以 Pulsar 為例（使用 zookeeper），看看在這類大型分布式系統(tǒng)中是如何處理負載均衡的。
再開始之前先明確下負載均衡大體上會做哪些事情。
- 首先上報自己節(jié)點的負載數(shù)據(jù)
- Leader 節(jié)點需要定時收集所有節(jié)點的負載數(shù)據(jù)。
- CPU、堆內(nèi)存、堆外內(nèi)存等通用數(shù)據(jù)的使用量
- 流出、流入流量
- 一些系統(tǒng)特有的數(shù)據(jù)，比如在 Pulsar 中就是：
- 每個 broker 中的 topic、consumer、producer、bundle 等數(shù)據(jù)。
- 這些負載數(shù)據(jù)中包括：
- 再由 leader 節(jié)點讀取到這些數(shù)據(jù)后選擇負載較高的節(jié)點，將數(shù)據(jù)遷移到負載較低的節(jié)點。
以上就是一個完整的負載均衡的流程，下面我們依次看看在 Pulsar 中是如何實現(xiàn)這些邏輯的。
在 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， 如果出現(xiàn)異常那就會使用 SimpleLoadManagerImpl 作為兜底。
他們兩個的區(qū)別是 ModularLoadManagerImpl 的功能更全，可以做更為細致的負載策略。
接下來以默認的 ModularLoadManagerImpl 為例講解上述的流程。
上報負載數(shù)據(jù)
在負載均衡器啟動的時候就會收集節(jié)點數(shù)據(jù)然后進行上報：
public void start() throws PulsarServerException {
try {
String brokerId = pulsar.getBrokerId();
brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + brokerId;
// 收集本地負載數(shù)據(jù)
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);
}
}首先獲取到當(dāng)前 broker 的 Id 然后拼接一個 zookeeper 節(jié)點的路徑，將生成的 localData 上傳到 zookeeper 中。
// 存放 broker 的節(jié)點信息
ls /loadbalance/brokers
[broker-1:8080, broker-2:8080]
// 根據(jù)節(jié)點信息查詢負載數(shù)據(jù)
get /loadbalance/brokers/broker-1:8080上報的數(shù)據(jù)：
{"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"}采集數(shù)據(jù)
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 和 堆外內(nèi)存的數(shù)據(jù)。
收集所有節(jié)點數(shù)據(jù)
作為 leader 節(jié)點還需要收集所有節(jié)點的負載數(shù)據(jù)，然后根據(jù)一些規(guī)則選擇將負載較高的節(jié)點移動到負債較低的節(jié)點中。
private void updateAllBrokerData() {
// 從 zookeeper 中獲取所有節(jié)點
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);
// 依次讀取各個節(jié)點的負載數(shù)據(jù)
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.
// 將數(shù)據(jù)寫入到本地緩存
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 的節(jié)點中獲取到所有的 broker 列表（broker 會在啟動時將自身的信息注冊到 zookeeper 中。）
然后依次讀取各自節(jié)點的負載數(shù)據(jù)，也就是在負載均衡器啟動的時候上報的數(shù)據(jù)。
篩選出所有 broker 中需要 unload 的 bundle
在 Pulsar 中 topic 是最核心的概念，而為了方便管理大量 topic，提出了一個 Bundle 的概念；Bundle 是一批 topic 的集合，管理 Bundle 自然會比 topic 更佳容易。
所以在 Pulsar 中做負載均衡最主要的就是將負載較高節(jié)點中的 bundle 轉(zhuǎn)移到低負載的 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);
}
}
}負載均衡器在啟動的時候就會查詢所有節(jié)點的數(shù)據(jù)，然后寫入到 brokerDataMap 中。
同時也會注冊相關(guān)的 zookeeper 事件，當(dāng)注冊的節(jié)點發(fā)生變化時（一般是新增或者刪減了 broker 節(jié)點）就會更新內(nèi)存中緩存的負載數(shù)據(jù)。
之后 leader 節(jié)點會定期調(diào)用 org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl#doLoadShedding 函數(shù)查詢哪些數(shù)據(jù)需要卸載，然后進行重新負載。
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);最核心的就是調(diào)用這個 findBundlesForUnloading 函數(shù)，會返回需要卸載 bundle 集合，最終會遍歷這個集合調(diào)用 admin API 進行卸載和重平衡。
而這個函數(shù)會有多種實現(xiàn)，本質(zhì)上就是根據(jù)傳入的各個節(jié)點的負載數(shù)據(jù)，然后根據(jù)自定義的規(guī)則返回一批需要卸載的數(shù)據(jù)。
以默認的 org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder 規(guī)則為例：
它是根據(jù)帶寬、內(nèi)存、流量等各個指標的權(quán)重算出每個節(jié)點的負載值，之后為整個集群計算出一個平均負載值。
以上圖為例：超過 ShedBundles 的數(shù)據(jù)就需要被卸載掉，然后轉(zhuǎn)移到低負載的節(jié)點中。
所以最左邊節(jié)點和超出的 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();
}
});
}從代碼里看的出來就是在一個備選集合中根據(jù)各種閾值和判斷條件篩選出需要卸載的 bundle。
而 SimpleLoadManagerImpl 的實現(xiàn)如下：
synchronized (currentLoadReports) {
for (Map.Entry<ResourceUnit, LoadReport> entry : currentLoadReports.entrySet()) {
ResourceUnit overloadedRU = entry.getKey();
LoadReport lr = entry.getValue();
// 所有數(shù)據(jù)做一個簡單的篩選，超過閾值的數(shù)據(jù)需要被 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;
}就是很簡單的通過將判斷節(jié)點的負載是否超過了閾值 isAboveLoadLevel，然后做一個簡單的排序就返回了。
從這里也看得出來 SimpleLoadManagerImpl 和 ModularLoadManager 的區(qū)別，SimpleLoadManagerImpl 更簡單，并沒有提供多個 doLoadShedding 的篩選實現(xiàn)。
總結(jié)
總的來說對于無狀態(tài)的服務(wù)來說，理論上我們只需要做好負載算法即可（輪訓(xùn)、一致性哈希、低負載優(yōu)先等）就可以很好的平衡各個節(jié)點之間的負載。
而對于有狀態(tài)的服務(wù)來說，負載均衡就是將負載較高節(jié)點中的數(shù)據(jù)轉(zhuǎn)移到負載低的節(jié)點中。
其中的關(guān)鍵就是需要存儲各個節(jié)點的負載數(shù)據(jù)（業(yè)界常用的是存儲到 zookeeper 中），然后再由一個 leader 節(jié)點從這些節(jié)點中根據(jù)某種負載算法選擇出負載較高的節(jié)點以及負載較低的節(jié)點，最終把數(shù)據(jù)遷移過去即可。
