大家好,我是君哥。

在講解 Kafka的副本同步限流機(jī)制三部曲(源碼篇) 第二篇(原理篇) 之前

我想先講解一下 Kafka中的數(shù)據(jù)采集和統(tǒng)計(jì)機(jī)制 當(dāng)你了解這個(gè)機(jī)制之后才會(huì)更容易理解限流機(jī)制 圖片

你會(huì)不會(huì)好奇,kafka監(jiān)控中,那些數(shù)據(jù)都是怎么計(jì)算出來(lái)的 比如下圖這些指標(biāo)

LogiKM監(jiān)控圖

這些數(shù)據(jù)都是通過(guò)Jmx獲取的kafka監(jiān)控指標(biāo), 那么我們今天來(lái)探討一下,這些指標(biāo)都是怎么被計(jì)算出來(lái)的

在開(kāi)始分析之前,我們可以 自己思考一下

如果讓你統(tǒng)計(jì)前一分鐘內(nèi)的流速,你會(huì)怎么統(tǒng)計(jì)才能夠讓數(shù)字更加精確呢？

我相信你腦海中肯定出現(xiàn)了一個(gè)詞：滑動(dòng)窗口

在kafka的數(shù)據(jù)采樣和統(tǒng)計(jì)中,也是用了這個(gè)方法, 通過(guò)多個(gè)樣本Sample進(jìn)行采樣,并合并統(tǒng)計(jì)

當(dāng)然這一個(gè)過(guò)程少不了滑動(dòng)窗口的影子

采集和統(tǒng)計(jì)類(lèi)圖

我們先看下整個(gè)Kafka的數(shù)據(jù)采集和統(tǒng)計(jì)機(jī)制的類(lèi)圖

數(shù)據(jù)采集和統(tǒng)計(jì)全類(lèi)圖

看著整個(gè)類(lèi)圖好像很復(fù)雜,但是最核心的就是兩個(gè)Interface接口

Measurable: 可測(cè)量的、可統(tǒng)計(jì)的 Interface。這個(gè)Interface 有一個(gè)方法, 專(zhuān)門(mén)用來(lái)計(jì)算需要被統(tǒng)計(jì)的值的

/**
* 測(cè)量這個(gè)數(shù)量并將結(jié)果作為雙精度返回
* 參數(shù)：
* config – 此指標(biāo)的配置
* now – 進(jìn)行測(cè)量的 POSIX 時(shí)間（以毫秒為單位）
* 返回：
* 測(cè)量值
*/
double measure(MetricConfig config, long now);

比如說(shuō)返回 近一分鐘的bytesIn

Stat: 記錄數(shù)據(jù), 上面的是統(tǒng)計(jì),但是統(tǒng)計(jì)需要數(shù)據(jù)來(lái)支撐, 這個(gè)Interface就是用來(lái)做記錄的,這個(gè)Interface有一個(gè)方法

/**
* 記錄給定的值
* 參數(shù)：
* config – 用于該指標(biāo)的配置
* value – 要記錄的值
* timeMs – 此值發(fā)生的 POSIX 時(shí)間（以毫秒為單位）
*/
void record(MetricConfig config, double value, long timeMs);

有了這兩個(gè)接口,就基本上可以記錄數(shù)據(jù)和數(shù)據(jù)統(tǒng)計(jì)了

當(dāng)然這兩個(gè)接口都有一個(gè) MetricConfig 對(duì)象

MetricConfig

這是一個(gè)統(tǒng)計(jì)配置類(lèi), 主要是定義采樣的樣本數(shù)、單個(gè)樣本的時(shí)間窗口大小、單個(gè)樣本的事件窗口大小、限流機(jī)制有了這樣一個(gè)配置了,就可以自由定義時(shí)間窗口的大小,和采樣的樣本數(shù)之類(lèi)的影響最終數(shù)據(jù)精度的變量。

這里我需要對(duì)兩個(gè)參數(shù)重點(diǎn)說(shuō)明一下

單個(gè)樣本的時(shí)間窗口大小: 當(dāng)前記錄時(shí)間 - 當(dāng)前樣本的開(kāi)始時(shí)間 >= 此值  則需要使用下一個(gè)樣本。單個(gè)樣本的事件窗口大小: 當(dāng)前樣本窗口時(shí)間次數(shù) >= 此值  則需要使用下一個(gè)樣本

在整個(gè)統(tǒng)計(jì)中,不一定是按照時(shí)間窗口來(lái)統(tǒng)計(jì)的, 也可以按照事件窗口來(lái)統(tǒng)計(jì), 具體按照不同需求選擇配置

好了,大家腦海里面已經(jīng)有了最基本的概念了,我們接下來(lái)就以一個(gè)kafka內(nèi)部經(jīng)常使用的 SampledStat 記錄和統(tǒng)計(jì)的抽象類(lèi)來(lái)好好的深入分析理解一下。

SampledStat 樣本記錄統(tǒng)計(jì)抽象類(lèi)

這個(gè)記錄統(tǒng)計(jì)抽象類(lèi),是按照采樣的形式來(lái)計(jì)算的。里面使用了一個(gè)或者多個(gè)樣本進(jìn)行采樣統(tǒng)計(jì) List<Sample> samples; 當(dāng)前使用的樣本: current樣本初始化的值: initialValue

SampledStat : 實(shí)現(xiàn)了MeasurableStat 的抽象類(lèi),說(shuō)明它又能采集記錄數(shù)據(jù),又能統(tǒng)計(jì)分析數(shù)據(jù)

當(dāng)然它自身也定義了有兩個(gè)抽象方法

/** 更新具體樣本的數(shù)值 (單個(gè)樣本)**/
 protected abstract void update(Sample sample, MetricConfig config, double value, long timeMs);

/**組合所有樣本的數(shù)據(jù) 來(lái)統(tǒng)計(jì)出想要的數(shù)據(jù) **/
 public abstract double combine(List<Sample> samples, MetricConfig config, long now);

SampledStat圖形化展示

如上圖所示, 是一個(gè)SampledStat 的圖形化展示, 其中定義了 若干個(gè)樣本 Sample

記錄數(shù)據(jù)

@Override
   public void record(MetricConfig config, double value, long timeMs) {
       Sample sample = current(timeMs);
       if (sample.isComplete(timeMs, config))
           sample = advance(config, timeMs);
       update(sample, config, value, timeMs);
       sample.eventCount += 1;
   }

1. 獲取當(dāng)前的Sample號(hào),如果沒(méi)有則創(chuàng)建一個(gè)新的Sample,  創(chuàng)建的時(shí)候設(shè)置 初始化值 和 Sample起始時(shí)間(當(dāng)前時(shí)間) ,并保存到樣品列表里面
2. 判斷這個(gè)Sample是否完成(超過(guò)窗口期),判斷的邏輯是 當(dāng)前時(shí)間 - 當(dāng)前Sample的開(kāi)始時(shí)間 >= 配置的時(shí)間窗口值 或者 事件總數(shù) >= 配置的事件窗口值

/** 當(dāng)前時(shí)間 - 當(dāng)前Sample的開(kāi)始時(shí)間 >= 配置的時(shí)間窗口值 或者  事件總數(shù) >= 配置的事件窗口值 **/
 public boolean isComplete(long timeMs, MetricConfig config) {
           return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
       }

3.如果這個(gè)Sample已經(jīng)完成(超過(guò)窗口期), 則開(kāi)始選擇下一個(gè)窗口,如果下一個(gè)還沒(méi)創(chuàng)建則創(chuàng)建新的,如果下一個(gè)已經(jīng)存在,則重置這個(gè)Sample

4.拿到最終要使用的Sample后, 將數(shù)據(jù)記錄到這個(gè)Sample中。具體怎么記錄是讓具體的實(shí)現(xiàn)類(lèi)來(lái)實(shí)現(xiàn)的,因?yàn)橄胍罱K統(tǒng)計(jì)的數(shù)據(jù)可以不一樣,比如你只想記錄Sample中的最大值,那么更新的時(shí)候判斷是不是比之前的值大則更新,如果你想統(tǒng)計(jì)平均值,那么這里就讓單個(gè)Sample中所有的值累加（最終會(huì) 除以 Sample數(shù)量 求平均數(shù)的）

5.記錄事件次數(shù)+1。

記錄數(shù)據(jù)的展示圖

統(tǒng)計(jì)數(shù)據(jù)

/** 測(cè)量  統(tǒng)計(jì) 數(shù)據(jù)**/
   @Override
   public double measure(MetricConfig config, long now) {
       // 重置過(guò)期樣本
       purgeObsoleteSamples(config, now);
       // 組合所有樣本數(shù)據(jù),并展示最終統(tǒng)計(jì)數(shù)據(jù),具體實(shí)現(xiàn)類(lèi)來(lái)實(shí)現(xiàn)該方法
       return combine(this.samples, config, now);
   }

先重置 過(guò)期樣本 , 過(guò)期樣本的意思是：當(dāng)前時(shí)間 - 每個(gè)樣本的起始事件 > 樣本數(shù)量 * 每個(gè)樣本的窗口時(shí)間 ; 就是滑動(dòng)窗口的概念,只統(tǒng)計(jì)這個(gè)滑動(dòng)窗口的樣本數(shù)據(jù), 過(guò)期的樣本數(shù)據(jù)會(huì)被重置(過(guò)期數(shù)據(jù)不采納), 如下圖所示

滑動(dòng)窗口重置過(guò)期數(shù)據(jù)

組合所有樣本數(shù)據(jù)并進(jìn)行不同維度的統(tǒng)計(jì)并返回?cái)?shù)值, 因?yàn)椴煌瑘?chǎng)景想要得到的數(shù)據(jù)不同，所以這個(gè)只是一個(gè)抽象方法,需要實(shí)現(xiàn)類(lèi)來(lái)實(shí)現(xiàn)這個(gè)計(jì)算邏輯,比如如果是計(jì)算平均值 Avg, 它的計(jì)算邏輯就是把所有的樣本數(shù)據(jù)值累加并除以累積的次數(shù)

那我們?cè)賮?lái)看看不同的統(tǒng)計(jì)實(shí)現(xiàn)類(lèi)

Avg 計(jì)算平均值

一個(gè)簡(jiǎn)單的SampledStat實(shí)現(xiàn)類(lèi) 它統(tǒng)計(jì)所有樣本最終的平均值 每個(gè)樣本都會(huì)累加每一次的記錄值， 最后把所有樣本數(shù)據(jù)疊加 / 總共記錄的次數(shù)

在這里插入圖片描述

Max 計(jì)算最大值

每個(gè)樣本都保存這個(gè)樣本的最大值, 然后最后再對(duì)比所有樣本值的最大值

在這里插入圖片描述

WindowedSum 所有樣本窗口總和值

每個(gè)樣本累積每一次的記錄值, 統(tǒng)計(jì)的時(shí)候 把所有樣本的累計(jì)值 再累積返回

在這里插入圖片描述

Rate 樣本記錄統(tǒng)計(jì)求速率

Rate 也是實(shí)現(xiàn)了 MeasurableStat接口的,說(shuō)明 它也有 記錄record 和 統(tǒng)計(jì) measure 的方法, 實(shí)際上這個(gè)類(lèi) 是一個(gè)組合類(lèi) ，里面組合了 SampledStat 和TimeUnit unit ,這不是很明顯了么, SampledStat負(fù)責(zé)記錄和統(tǒng)計(jì), 得到的數(shù)據(jù) 跟時(shí)間TimeUnit做一下處理就得出來(lái)速率了, 比如SampledStat的實(shí)現(xiàn)類(lèi)AVG可以算出來(lái) 被統(tǒng)計(jì)的 評(píng)價(jià)值, 但是如果我們?cè)俪?一個(gè)時(shí)間維度, 是不是就可以得出 平均速率 了

如何計(jì)算統(tǒng)計(jì)的有效時(shí)間呢

這個(gè)有效時(shí)間 的計(jì)算會(huì)影響著最終速率的結(jié)果

public long windowSize(MetricConfig config, long now) {
       // 將過(guò)期的樣本給重置掉
       stat.purgeObsoleteSamples(config, now);
       // 總共運(yùn)行的時(shí)候 = 當(dāng)前時(shí)間 - 最早的樣本的開(kāi)始時(shí)間
       long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
       // 總時(shí)間/單個(gè)創(chuàng)建時(shí)間 = 多少個(gè)完整的窗口時(shí)間
       int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
       int minFullWindows = config.samples() - 1;
       // If the available windows are less than the minimum required, add the difference to the totalElapsedTime
       if (numFullWindows < minFullWindows)
           totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
       return totalElapsedTimeMs;
   }

這是Rate的有效時(shí)間的計(jì)算邏輯,當(dāng)然Rate 還有一個(gè)子類(lèi)是 SampleRate

SampleRate的窗口Size計(jì)算邏輯

這個(gè)子類(lèi),將 有效時(shí)間的計(jì)算邏輯改的更簡(jiǎn)單, 如果運(yùn)行時(shí)間<一個(gè)樣本窗口的時(shí)間 則他的運(yùn)行時(shí)間就是單個(gè)樣本的窗口時(shí)間, 否則就直接用這個(gè)運(yùn)行的時(shí)間, 這個(gè)計(jì)算邏輯更簡(jiǎn)單 它跟Rate的區(qū)別就是, 不考慮采樣的時(shí)間是否足夠多,我們用圖來(lái)簡(jiǎn)單描述一下

SampleRate

SampleRate 速率邏輯

Rate

Rate 速率邏輯

Meter 包含速率和累積總指標(biāo)的復(fù)合統(tǒng)計(jì)數(shù)據(jù)

這是一個(gè)CompoundStat的實(shí)現(xiàn)類(lèi), 說(shuō)明它是一個(gè)復(fù)合統(tǒng)計(jì), 可以統(tǒng)計(jì)很多指標(biāo)在這里面 它包含速率指標(biāo)和累積總指標(biāo)的復(fù)合統(tǒng)計(jì)數(shù)據(jù)

底層實(shí)現(xiàn)的邏輯還是上面講解過(guò)的

副本Fetch流量的速率統(tǒng)計(jì) 案例分析

我們知道 在分區(qū)副本重分配過(guò)程中,有一個(gè)限流機(jī)制,就是指定某個(gè)限流值,副本同步過(guò)程不能超過(guò)這個(gè)閾值。做限流,那么肯定首先就需要統(tǒng)計(jì) 副本同步 的流速；那么上面我們講了這么多,你應(yīng)該很容易能夠想到如果統(tǒng)計(jì)了吧？流速  bytes/s , 統(tǒng)計(jì)一秒鐘同步了多少流量, 那么我們可以把樣本窗口設(shè)置為 1s,然后多設(shè)置幾個(gè)樣本窗口求平均值。

接下來(lái)我們看看 Kafka是怎么統(tǒng)計(jì)的, 首先找到記錄 Follower Fetch 副本流量的地方如下

ReplicaFetcherThread#processPartitionData
if(quota.isThrottled(topicPartition))
 quota.record(records.sizeInBytes)

設(shè)置時(shí)間窗口配置

這里設(shè)置的timeWindowMs 單個(gè)樣本窗口時(shí)間= 1 snumQuotaSamples 樣本數(shù) = 11 當(dāng)然這些都是可以配置的

查看使用了哪個(gè)實(shí)現(xiàn)類(lèi)

我們可以看到最終是使用了 SampleRate 來(lái)統(tǒng)計(jì)流量 !

Gauge 瞬時(shí)讀數(shù)的指標(biāo)

上面我們起始是主要講解了Measurable接口, 它的父類(lèi)是MetricValueProvider<Double> ,它沒(méi)有方法,只是定義,當(dāng)還有一個(gè)子接口是 Gauge ,它并不是上面那種采樣的形式來(lái)統(tǒng)計(jì)數(shù)據(jù), 它返回的是當(dāng)前的值, 瞬時(shí)值它提供的方法是 value() ， Measurable提供的是measure()

這個(gè)在kafka中使用場(chǎng)景很少,就不詳細(xì)介紹了。

好了,這一篇我們主要講解了一下 Kafka中的數(shù)據(jù)采集和統(tǒng)計(jì)機(jī)制

那么 接下來(lái)下一篇,我們來(lái)聊聊 Kafka的監(jiān)控機(jī)制, 如何把這些采集

到的信息給保存起來(lái)并對(duì)外提供!!!