直屏旗艦來了 iQOO 12和K70 Pro同臺競技
旗艦機(jī)基本上使用的都是雙曲面屏幕，這就讓很多喜歡直屏的愛好者在苦等一款直屏旗艦，這次，你們等到了。據(jù)博主數(shù)碼閑聊站帶來的最新爆料稱，Redmi下代旗艦K70 Pro和iQOO 12兩款手
Pulsar 有提供一個查詢 Broker 負(fù)載的接口：
/** * 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 的負(fù)載數(shù)據(jù)，比如 CPU、內(nèi)存、流量之類的數(shù)據(jù)。
我目前碰到的問題是目前會遇到部分節(jié)點的負(fù)債不平衡，導(dǎo)致資源占用不均衡，所以想要手動查詢所有節(jié)點的負(fù)載數(shù)據(jù)，然后人工進(jìn)行負(fù)載。
理論上這些數(shù)據(jù)是在運(yùn)行時實時計算的數(shù)據(jù)，如果對于單機(jī)的倒還好說，每次請求這個接口直接實時計算一次就可以了。
但對于集群的服務(wù)來說會有多個節(jié)點，目前 Pulsar 提供的這個接口只能查詢指定節(jié)點的負(fù)載數(shù)據(jù)，也就是說每次得傳入目標(biāo)節(jié)點的 IP 和端口。
提供的命令行工具其實就是直接從 zookeeper 中查詢的數(shù)據(jù)。
在分布式系統(tǒng)中需要一個集中的組件來管理各種數(shù)據(jù)，比如：
可以提供該功能的組件其實也不少：
Zookeeper 是老牌的分布式協(xié)調(diào)組件，可以做 leader 選舉、配置中心、分布式鎖、服務(wù)注冊與發(fā)現(xiàn)等功能。
在許多中間件和系統(tǒng)中都有應(yīng)用，比如：
etcd 的功能與 zookeeper 類似，可以用作服務(wù)注冊發(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)中是如何處理負(fù)載均衡的。
再開始之前先明確下負(fù)載均衡大體上會做哪些事情。
以上就是一個完整的負(fù)載均衡的流程，下面我們依次看看在 Pulsar 中是如何實現(xiàn)這些邏輯的。
在 Pulsar 中提供了多種負(fù)載均衡策略，以下是加載負(fù)載均衡器的邏輯：
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); }默認(rèn)使用的是 ModularLoadManagerImpl， 如果出現(xiàn)異常那就會使用 SimpleLoadManagerImpl 作為兜底。
他們兩個的區(qū)別是 ModularLoadManagerImpl 的功能更全，可以做更為細(xì)致的負(fù)載策略。
接下來以默認(rèn)的 ModularLoadManagerImpl 為例講解上述的流程。
在負(fù)載均衡器啟動的時候就會收集節(jié)點數(shù)據(jù)然后進(jìn)行上報：
public void start() throws PulsarServerException { try { String brokerId = pulsar.getBrokerId(); brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + brokerId; // 收集本地負(fù)載數(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é)點信息查詢負(fù)載數(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; }會在運(yùn)行時獲取一些 JVM 和 堆外內(nèi)存的數(shù)據(jù)。
作為 leader 節(jié)點還需要收集所有節(jié)點的負(fù)載數(shù)據(jù)，然后根據(jù)一些規(guī)則選擇將負(fù)載較高的節(jié)點移動到負(fù)債較低的節(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é)點的負(fù)載數(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é)點的負(fù)載數(shù)據(jù)，也就是在負(fù)載均衡器啟動的時候上報的數(shù)據(jù)。
在 Pulsar 中 topic 是最核心的概念，而為了方便管理大量 topic，提出了一個 Bundle 的概念；Bundle 是一批 topic 的集合，管理 Bundle 自然會比 topic 更佳容易。
所以在 Pulsar 中做負(fù)載均衡最主要的就是將負(fù)載較高節(jié)點中的 bundle 轉(zhuǎn)移到低負(fù)載的 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); } } }負(fù)載均衡器在啟動的時候就會查詢所有節(jié)點的數(shù)據(jù)，然后寫入到 brokerDataMap 中。
同時也會注冊相關(guān)的 zookeeper 事件，當(dāng)注冊的節(jié)點發(fā)生變化時（一般是新增或者刪減了 broker 節(jié)點）就會更新內(nèi)存中緩存的負(fù)載數(shù)據(jù)。
之后 leader 節(jié)點會定期調(diào)用 org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl#doLoadShedding 函數(shù)查詢哪些數(shù)據(jù)需要卸載，然后進(jìn)行重新負(fù)載。
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);最核心的就是調(diào)用這個 findBundlesForUnloading 函數(shù)，會返回需要卸載 bundle 集合，最終會遍歷這個集合調(diào)用 admin API 進(jìn)行卸載和重平衡。
而這個函數(shù)會有多種實現(xiàn)，本質(zhì)上就是根據(jù)傳入的各個節(jié)點的負(fù)載數(shù)據(jù)，然后根據(jù)自定義的規(guī)則返回一批需要卸載的數(shù)據(jù)。
以默認(rèn)的 org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder 規(guī)則為例：
它是根據(jù)帶寬、內(nèi)存、流量等各個指標(biāo)的權(quán)重算出每個節(jié)點的負(fù)載值，之后為整個集群計算出一個平均負(fù)載值。
以上圖為例：超過 ShedBundles 的數(shù)據(jù)就需要被卸載掉，然后轉(zhuǎn)移到低負(fù)載的節(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é)點的負(fù)載是否超過了閾值 isAboveLoadLevel，然后做一個簡單的排序就返回了。
從這里也看得出來 SimpleLoadManagerImpl 和 ModularLoadManager 的區(qū)別，SimpleLoadManagerImpl 更簡單，并沒有提供多個 doLoadShedding 的篩選實現(xiàn)。
總的來說對于無狀態(tài)的服務(wù)來說，理論上我們只需要做好負(fù)載算法即可（輪訓(xùn)、一致性哈希、低負(fù)載優(yōu)先等）就可以很好的平衡各個節(jié)點之間的負(fù)載。
而對于有狀態(tài)的服務(wù)來說，負(fù)載均衡就是將負(fù)載較高節(jié)點中的數(shù)據(jù)轉(zhuǎn)移到負(fù)載低的節(jié)點中。
其中的關(guān)鍵就是需要存儲各個節(jié)點的負(fù)載數(shù)據(jù)（業(yè)界常用的是存儲到 zookeeper 中），然后再由一個 leader 節(jié)點從這些節(jié)點中根據(jù)某種負(fù)載算法選擇出負(fù)載較高的節(jié)點以及負(fù)載較低的節(jié)點，最終把數(shù)據(jù)遷移過去即可。
本文鏈接：http://m.www897cc.com/showinfo-26-101276-0.html負(fù)載均衡技術(shù)全解析：Pulsar 分布式系統(tǒng)的優(yōu)秀實踐
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請聯(lián)系：2376512515@qq.com 津ICP備18001702號
津公網(wǎng)安備 12010102000574號