前言

前段時(shí)間我們?cè)谏?jí) Pulsar 版本的時(shí)候發(fā)現(xiàn)升級(jí)后最后一個(gè)節(jié)點(diǎn)始終沒(méi)有流量。

雖然對(duì)業(yè)務(wù)使用沒(méi)有任何影響，但負(fù)載不均會(huì)導(dǎo)致資源的浪費(fèi)。

和同事溝通后得知之前的升級(jí)也會(huì)出現(xiàn)這樣的情況，最終還是人工調(diào)用 Pulsar 的 admin API 完成的負(fù)載均衡。

這個(gè)問(wèn)題我嘗試在 Google 和 Pulsar 社區(qū)都沒(méi)有找到類似的，不知道是大家都沒(méi)碰到還是很少升級(jí)集群。

我之前所在的公司就是一個(gè)版本走到黑

Pulsar 負(fù)載均衡原理

當(dāng)一個(gè)集群可以水平擴(kuò)展后負(fù)載均衡就顯得非常重要，根本目的是為了讓每個(gè)提供服務(wù)的節(jié)點(diǎn)都能均勻的處理請(qǐng)求，不然擴(kuò)容就沒(méi)有意義了。

在分析這個(gè)問(wèn)題的原因之前我們先看看 Pulsar 負(fù)載均衡的實(shí)現(xiàn)方案。

# Enable load balancer
loadBalancerEnabled=true

我們可以通過(guò)這個(gè) broker 的這個(gè)配置來(lái)控制負(fù)載均衡器的開(kāi)關(guān)，默認(rèn)是打開(kāi)。

但具體使用哪個(gè)實(shí)現(xiàn)類來(lái)作為負(fù)載均衡器也可以在配置文件中指定：

# Name of load manager to use
loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl

默認(rèn)使用的是 ModularLoadManagerImpl。

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);
    }

當(dāng) broker 啟動(dòng)時(shí)會(huì)從配置文件中讀取配置進(jìn)行加載，如果加載失敗會(huì)使用 SimpleLoadManagerImpl 作為兜底策略。

當(dāng) broker 是一個(gè)集群時(shí)，只有 leader 節(jié)點(diǎn)的 broker 才會(huì)執(zhí)行負(fù)載均衡器的邏輯。

Leader 選舉是通過(guò) Zookeeper 實(shí)現(xiàn)的。

默然情況下成為 Leader 節(jié)點(diǎn)的 broker 會(huì)每分鐘讀取各個(gè) broker 的數(shù)據(jù)來(lái)判斷是否有節(jié)點(diǎn)負(fù)載過(guò)高需要做重平衡。

而是否重平衡的判斷依據(jù)是由 org.apache.pulsar.broker.loadbalance.LoadSheddingStrategy 接口提供的，它其實(shí)只有一個(gè)函數(shù)：

public interface LoadSheddingStrategy {

    /**
     * Recommend that all of the returned bundles be unloaded.
     * @return A map from all selected bundles to the brokers on which they reside.
     */
    Multimap<String, String> findBundlesForUnloading(LoadData loadData, ServiceConfiguration conf);
}

根據(jù)所有 broker 的負(fù)載信息計(jì)算出一個(gè)需要被 unload 的 broker 以及 bundle。

這里解釋下 unload 和 bundle 的概念：

• bundle 是一批 topic 的抽象，將 bundle 和 broker 進(jìn)行關(guān)聯(lián)后客戶端才能知道應(yīng)當(dāng)連接哪個(gè) broker；而不是直接將 topic 與 broker 綁定，這樣才能實(shí)現(xiàn)海量 topic 的管理。
• unload 則是將已經(jīng)與 broker 綁定的 bundle 手動(dòng)解綁，從而觸發(fā)負(fù)載均衡器選擇一臺(tái)合適的 broker 重新進(jìn)行綁定；通常是整個(gè)集群負(fù)載不均的時(shí)候觸發(fā)。

ThresholdShedder 原理

LoadSheddingStrategy 接口目前有三個(gè)實(shí)現(xiàn)，這里以官方默認(rèn)的 ThresholdShedder 為例：

它的實(shí)現(xiàn)算法是根據(jù)帶寬、內(nèi)存、流量等各個(gè)指標(biāo)的權(quán)重算出每個(gè)節(jié)點(diǎn)的負(fù)載值，之后為整個(gè)集群計(jì)算出一個(gè)平均負(fù)載值。

# 閾值
loadBalancerBrokerThresholdShedderPercentage=10

當(dāng)集群中有某個(gè)節(jié)點(diǎn)的負(fù)載值超過(guò)平均負(fù)載值達(dá)到一定程度（可配置的閾值）時(shí)，就會(huì)觸發(fā) unload，以上圖為例就會(huì)將最左邊節(jié)點(diǎn)中紅色部分的 bundle 卸載掉，然后再重新計(jì)算一個(gè)合適的 broker 進(jìn)行綁定。

閾值存在的目的是為了避免頻繁的 unload，從而影響客戶端的連接。

問(wèn)題原因

當(dāng)某些 topic 的流量突然爆增的時(shí)候這種負(fù)載策略確實(shí)可以處理的很好，但在我們集群升級(jí)的情況就不一定了。

假設(shè)我這里有三個(gè)節(jié)點(diǎn)：

• broker0
• broker1
• broker2

集群升級(jí)時(shí)會(huì)從 broker2->0 進(jìn)行鏡像替換重啟，假設(shè)在升級(jí)前每個(gè) broker 的負(fù)載值都是 10。

• 重啟 broker2 時(shí)，它所綁定的 bundle 被 broker0/1 接管。
• 升級(jí) broker1 時(shí)，它所綁定的 bundle 又被 broker0/2 接管。
• 最后升級(jí) broker0, 它所綁定的 bundle 會(huì)被broker1/2 接管。

只要在這之后沒(méi)有發(fā)生流量激增到觸發(fā)負(fù)載的閾值，那么當(dāng)前的負(fù)載情況就會(huì)一直保留下去，也就是 broker0 會(huì)一直沒(méi)有流量。

經(jīng)過(guò)我反復(fù)測(cè)試，現(xiàn)象也確實(shí)如此。

./pulsar-perf monitor-brokers --connect-string pulsar-test-zookeeper:2181

通過(guò)這個(gè)工具也可以查看各個(gè)節(jié)點(diǎn)的負(fù)載情況

優(yōu)化方案

這種場(chǎng)景是當(dāng)前 ThresholdShedder 所沒(méi)有考慮到的，于是我在我們所使用的版本 2.10.3 的基礎(chǔ)上做了簡(jiǎn)單的優(yōu)化：

• 當(dāng)原有邏輯走完之后也沒(méi)有獲取需要需要卸載的 bundle，同時(shí)也存在一個(gè)負(fù)載極低的 broker 時(shí)(emptyBundle)，再觸發(fā)一次 bundle 查詢。
• 按照 broker 所綁定的數(shù)量排序，選擇一個(gè)數(shù)量最多的 broker 的 第一個(gè) bundle 進(jìn)行卸載。

修改后打包發(fā)布，再走一遍升級(jí)流程后整個(gè)集群負(fù)載就是均衡的了。

但其實(shí)這個(gè)方案并不嚴(yán)謹(jǐn)，第二步選擇的重點(diǎn)是篩選出負(fù)載最高的集群中負(fù)載最高的 bundle；這里只是簡(jiǎn)單的根據(jù)數(shù)量來(lái)判斷，并不夠準(zhǔn)確。

正當(dāng)我準(zhǔn)備持續(xù)優(yōu)化時(shí)，鬼使神差的我想看看 master 上有人修復(fù)這個(gè)問(wèn)題沒(méi)，結(jié)果一看還真有人修復(fù)了；只是還沒(méi)正式發(fā)版。

??https://github.com/apache/pulsar/pull/17456??

整體思路是類似的，只是篩選負(fù)載需要卸載 bundle 時(shí)是根據(jù) bundle 自身的流量來(lái)的，這樣會(huì)更加精準(zhǔn)。

總結(jié)

不過(guò)看社區(qū)的進(jìn)度等這個(gè)優(yōu)化最終能用還不知道得多久，于是我們就自己參考這個(gè)思路在管理臺(tái)做了類似的功能，當(dāng)升級(jí)后出現(xiàn)負(fù)載不均衡時(shí)人工觸發(fā)一個(gè)邏輯：

• 系統(tǒng)根據(jù)各個(gè)節(jié)點(diǎn)的負(fù)載情況計(jì)算出一個(gè)負(fù)載最高的節(jié)點(diǎn)和 bundle 在頁(yè)面上展示。
• 人工二次確認(rèn)是否要卸載，確認(rèn)無(wú)誤后進(jìn)行卸載。

本質(zhì)上只是將上述優(yōu)化的自動(dòng)負(fù)載流程改為人工處理了，經(jīng)過(guò)測(cè)試效果是一樣的。

Pulsar 整個(gè)項(xiàng)目其實(shí)非常龐大，有著幾十上百個(gè)模塊，哪怕每次我只改動(dòng)一行代碼準(zhǔn)備發(fā)布測(cè)試時(shí)都得經(jīng)過(guò)漫長(zhǎng)的編譯+ Docker鏡像打包+上傳私服這些流程，通常需要1~2個(gè)小時(shí)；但總的來(lái)說(shuō)收獲還是很大的，最近也在提一些 issue 和 PR，希望后面能更深入的參與進(jìn)社區(qū)。