Kafka Eagle 源碼解讀

2019-11-06 06:04:13

字體：大中小

來源：轉載

供稿：網友

http://www.cnblogs.com/smartloli/p/6490643.html

1.概述

　　在《Kafka 消息監控 - Kafka Eagle》一文中，簡單的介紹了 Kafka Eagle這款監控工具的作用，截圖預覽，以及使用詳情。今天筆者通過其源碼來解讀實現細節。目前該項目已托管于 Github 之上，作者編寫了使用手冊，告知使用者如何安裝，部署，啟動該系統。但對于實現的細節并未在參考手冊中詳細指出。這里，筆者通過本篇博文，來詳細解讀其實現細節。相關資料文獻地址如下所示：

Kafka Eagle 源碼地址Kafka Eagle 參考手冊Kafka Eagle 安裝包

2.內容

　　截止到版本 Kafka Eagle v1.1.1 支持監控0.8.2.x（存放消費信息于Zookeeper）以及 0.10.x（存放消費信息于Kafka的topic中）。對于前者，從Zookeeper中獲取消息信息，難度不大，編寫Zookeeper客戶端實現代碼即可，該版本在Zookeeper下的存儲結構樹如下圖所示：

對于實現細節，可使用ZkUtils工具類來獲取相關數據，以獲取消費信息為例，代碼如下所示：

/** Obtaining kafka consumer information from zookeeper. */ public Map<String, List<String>> getConsumers(String clusterAlias) { ZkClient zkc = zkPool.getZkClient(clusterAlias); Map<String, List<String>> consumers = new HashMap<String, List<String>>(); try { Seq<String> subConsumerPaths = ZkUtils.getChildren(zkc, CONSUMERS_PATH); List<String> groups = javaConversions.seqAsJavaList(subConsumerPaths); for (String group : groups) { String path = CONSUMERS_PATH + "/" + group + "/owners"; if (ZkUtils.pathExists(zkc, path)) { Seq<String> owners = ZkUtils.getChildren(zkc, path); List<String> ownersserialize = JavaConversions.seqAsJavaList(owners); consumers.put(group, ownersSerialize); } else { LOG.error("Consumer Path[" + path + "] is not exist."); } } } catch (Exception ex) { LOG.error(ex.getMessage()); } finally { if (zkc != null) { zkPool.release(clusterAlias, zkc); zkc = null; } } return consumers; } 復制代碼

其他監控信息可以按照Zookeeper中結構樹路徑獲取。如下圖所示：

然而，對于新版本，官方默認將消費信息遷移到Kafka的topic中，這樣原來的接口只能獲取topic，broker等信息，對于消費的信息，我們需要從kafka中一個叫__consumer_offsets的topic中獲取。為了兼容0.8.2.x版本的Kafka，這里在Kafka Eagle中另外啟動一個RpcServer來貢獻__consumer_offsets中的消費信息。消費__consumer_offsets這個topic時，需要指定該內部topic不暴露給consumer，將 exclude.internal.topics 設置為 false 即可。這樣我們通過一個 kafka.eagle.offset.storage 開關來控制系統獲取監控元數據的走向。獲取流程如下圖所示：

3.消費 Owner

　　當消費的信息存放于Zookeeper中，我們可以直接從consumer模塊下直接獲取對應的Owner，但是在Kafka的Topic中，我們需要編碼來間接的獲取。這里，我們需要知道 Kafka 的Owner的組成規則，其規則由 Group+ConusmerHostAddress+Timespan+UUID+PartitionId組成，實現細節可參考源碼，界面展示如下圖所示：

4.Kafka SQL

　　關于Kafka SQL，旨在使用SQL來快速可視化Topic的相關信息，目前 Kafka SQL 實現的功能包含有展示某一個Topic的Partition，Offset，以及其對應的消息記錄，若不加limit條件限制，默認展示該Topic下最新的5000條記錄，詳細實現細節，可參看源碼，預覽截圖如下所示：

查詢結果，如下圖所示：

5.多集群

　　Kafka Eagle 目前是支持多集群監控，所謂多集群，是指多個Zookeeper集群下的Kafka集群，通過切換session來管理不同的Zookeeper集群下的Kafka集群，細節參看源碼。管理界面如下圖所示：

6.總結

　　Kafka Eagle總體實現思路基本如上所述。針對，Kafka 0.10.x版本，Kafka Eagle監控部分模塊不展示的問題，這里在啟動 Kafka Eagle之前，默認啟動一個系統consumer來消費kafka.eagle該group下的__system.topic__，保證__consumer_offsets是有數據可供獲取的。