容量越大越不壞？24萬塊硬盤故障率報告公布 這些產品零故障
8月5日消息，云存儲服務商Backblaze發布了最新的硬盤故障率報告，年故障率有所上升。Backblaze發布的硬盤季度統計數據，其中包括故障率等重要方面。這些結
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 和端口。
提供的命令行工具其實就是直接從 zookeeper 中查詢的數據。
在分布式系統中需要一個集中的組件來管理各種數據，比如：
可以提供該功能的組件其實也不少：
Zookeeper 是老牌的分布式協調組件，可以做 leader 選舉、配置中心、分布式鎖、服務注冊與發現等功能。
在許多中間件和系統中都有應用，比如：
etcd 的功能與 zookeeper 類似，可以用作服務注冊發現，也可以作為 Key Value 鍵值對存儲系統；在 kubernetes 中扮演了巨大作用，經歷了各種考驗，穩定性已經非常可靠了。
Oxia 則是 StreamNative 開發的一個用于替換 Zookeeper 的中間件，功能也與 Zookeeper 類似；目前已經可以在 Pulsar 中替換 Zookeeper，只是還沒有大規模的使用。
下面以 Pulsar 為例（使用 zookeeper），看看在這類大型分布式系統中是如何處理負載均衡的。
再開始之前先明確下負載均衡大體上會做哪些事情。
以上就是一個完整的負載均衡的流程，下面我們依次看看在 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 中。）
然后依次讀取各自節點的負載數據，也就是在負載均衡器啟動的時候上報的數據。
在 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 節點從這些節點中根據某種負載算法選擇出負載較高的節點以及負載較低的節點，最終把數據遷移過去即可。
本文鏈接：http://www.www897cc.com/showinfo-26-101276-0.html負載均衡技術全解析：Pulsar 分布式系統的優秀實踐
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： B+樹層面查詢數據的全過程詳解
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號