這個包主要是與Kafka度量相關的。

一、KafkaTimer.scala 對代碼塊的運行進行計時。僅提供一個方法: timer——在運行傳入函數f的同時為期計時 二、KafkaMetricsConfig.scala 指定reporter類，以逗號分隔的reporter類，比如kafka.metrics.KafkaCSVMetricsReporter，這些類必須要在claasspath中指定。另外指定了度量的輪詢間隔，默認是10秒。 三、KafkaMetricsReporter.scala KafkaMetricsReporterMBean是基本trait。客戶如果要自己定制reporter的話除了要實現KafkaMetricsReporter之外還要實現這個MBean以滿足標準的MBean規(guī)范。該trait定義了3個抽象方法： 1. startReporter: 調用yammer的CsvReporter的start方法開啟reporter 2. stopReporter: 調用yammer的CsvReporter的shutdown方法關閉reporter 3. getMBeanName: 獲取MBean的名稱，格式為：kafka:type=kafka.metrics.KafkaCSVMetricsReporter 該scala文件還提供了第二個trait: KafkaMetricsReporter，該trait只有一個方法init，在KafkaCSVMetricsReporter.scala中實現了該init方法 最后，該scala還有一個object: KafkaMetricsReporter。該object維護了一個AtomicBoolean的ReporterStarted變量，標識該reporter是否已經啟動，并在啟動reporter的過程中充當鎖的作用。該object唯一的方法startReporters就是啟動MetricConfig中定義的所有reporter——具體方法是調用Utils.createObject方法通過反射機制創(chuàng)建所有reporter，并初始化每個reporter，最后將reporter注冊到MBean中。 四、KafkaCSVMetricsReporter.scala 該scala可以說Kafka默認提供的一個具體reporter。一共提供了兩個MBean：KafkaCSVMetricsReporterMBean和KafkaCSVMetricsReporter。我們先說第一個：KafkaCSVMetricsReporterMBean。該trait是一個私有trait，就是簡單地繼承了KafkaMetricsReporterMBean。 而第二個trait繼承了KafkaMetricsMBean和KafkaCSVMetricsReporterMBean，因而實現了前兩個trait的4個方法，分別是init, getMBeanName, startReporter和stopReporter。值得一提的是，在init方法中它會創(chuàng)建一個默認為kafka_metrics的子目錄，然后判斷是否kafka.csv.metrics.reporter.enabled設定為true，如果true的話調用startReporter方法開啟report，否則就直接結束。 五、KafkaMetricsGroup.scala 該scala定義了一個trait和一個object。就KafkaMetricsGroup trait而言，顧名思義，這應該是個metric組，定義了很多方法，主要的方法就是newGauge、newMeter、newHistogram和newTimer調用yammer框架的api創(chuàng)建不同的度量元對象。 而該scala的object： KafkaMetricsGroup分別為consumer、server、producer定義了各自的metric列表。當然還提供了remove方法刪除給定clientId的metric

轉載于:https://www.cnblogs.com/huxi2b/p/4401461.html

總結

以上是生活随笔為你收集整理的【原创】Kakfa metrics包源代码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。