對標蘋果的靈動島 華為帶來實況窗功能
繼蘋果的靈動島之后，華為也在今天正式推出了“實況窗”功能。據(jù)今天鴻蒙OS 4.0的現(xiàn)場演示顯示，華為的實況窗可以更高效的展現(xiàn)出實時通知，比如鎖屏上就能看到外賣、打車、銀行
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é)點的負債不平衡，導致資源占用不均衡，所以想要手動查詢所有節(jié)點的負載數(shù)據(jù)，然后人工進行負載。
理論上這些數(shù)據(jù)是在運行時實時計算的數(shù)據(jù)，如果對于單機的倒還好說，每次請求這個接口直接實時計算一次就可以了。
但對于集群的服務來說會有多個節(jié)點，目前 Pulsar 提供的這個接口只能查詢指定節(jié)點的負載數(shù)據(jù)，也就是說每次得傳入目標節(jié)點的 IP 和端口。
提供的命令行工具其實就是直接從 zookeeper 中查詢的數(shù)據(jù)。
在分布式系統(tǒng)中需要一個集中的組件來管理各種數(shù)據(jù)，比如：
可以提供該功能的組件其實也不少：
Zookeeper 是老牌的分布式協(xié)調組件，可以做 leader 選舉、配置中心、分布式鎖、服務注冊與發(fā)現(xiàn)等功能。
在許多中間件和系統(tǒng)中都有應用，比如：
etcd 的功能與 zookeeper 類似，可以用作服務注冊發(fā)現(xiàn)，也可以作為 Key Value 鍵值對存儲系統(tǒng)；在 kubernetes 中扮演了巨大作用，經(jīng)歷了各種考驗，穩(wěn)定性已經(jīng)非常可靠了。
Oxia 則是 StreamNative 開發(fā)的一個用于替換 Zookeeper 的中間件，功能也與 Zookeeper 類似；目前已經(jīng)可以在 Pulsar 中替換 Zookeeper，只是還沒有大規(guī)模的使用。
下面以 Pulsar 為例（使用 zookeeper），看看在這類大型分布式系統(tǒng)中是如何處理負載均衡的。
再開始之前先明確下負載均衡大體上會做哪些事情。
以上就是一個完整的負載均衡的流程，下面我們依次看看在 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 為例講解上述的流程。
在負載均衡器啟動的時候就會收集節(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); } }首先獲取到當前 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"}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ù)。
作為 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ù)。
在 Pulsar 中 topic 是最核心的概念，而為了方便管理大量 topic，提出了一個 Bundle 的概念；Bundle 是一批 topic 的集合，管理 Bundle 自然會比 topic 更佳容易。
所以在 Pulsar 中做負載均衡最主要的就是將負載較高節(jié)點中的 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); } } }負載均衡器在啟動的時候就會查詢所有節(jié)點的數(shù)據(jù)，然后寫入到 brokerDataMap 中。
同時也會注冊相關的 zookeeper 事件，當注冊的節(jié)點發(fā)生變化時（一般是新增或者刪減了 broker 節(jié)點）就會更新內(nèi)存中緩存的負載數(shù)據(jù)。
之后 leader 節(jié)點會定期調用 org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl#doLoadShedding 函數(shù)查詢哪些數(shù)據(jù)需要卸載，然后進行重新負載。
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);最核心的就是調用這個 findBundlesForUnloading 函數(shù)，會返回需要卸載 bundle 集合，最終會遍歷這個集合調用 admin API 進行卸載和重平衡。
而這個函數(shù)會有多種實現(xiàn)，本質上就是根據(jù)傳入的各個節(jié)點的負載數(shù)據(jù)，然后根據(jù)自定義的規(guī)則返回一批需要卸載的數(shù)據(jù)。
以默認的 org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder 規(guī)則為例：
它是根據(jù)帶寬、內(nèi)存、流量等各個指標的權重算出每個節(jié)點的負載值，之后為整個集群計算出一個平均負載值。
以上圖為例：超過 ShedBundles 的數(shù)據(jù)就需要被卸載掉，然后轉移到低負載的節(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)。
總的來說對于無狀態(tài)的服務來說，理論上我們只需要做好負載算法即可（輪訓、一致性哈希、低負載優(yōu)先等）就可以很好的平衡各個節(jié)點之間的負載。
而對于有狀態(tài)的服務來說，負載均衡就是將負載較高節(jié)點中的數(shù)據(jù)轉移到負載低的節(jié)點中。
其中的關鍵就是需要存儲各個節(jié)點的負載數(shù)據(jù)（業(yè)界常用的是存儲到 zookeeper 中），然后再由一個 leader 節(jié)點從這些節(jié)點中根據(jù)某種負載算法選擇出負載較高的節(jié)點以及負載較低的節(jié)點，最終把數(shù)據(jù)遷移過去即可。
本文鏈接：http://www.tebozhan.com/showinfo-26-101276-0.html負載均衡技術全解析：Pulsar 分布式系統(tǒng)的優(yōu)秀實踐
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯(lián)系：2376512515@qq.com 津ICP備18001702號
津公網(wǎng)安備 12010102000574號