第24講：通過 Kafka Eagle 實現對 Kafka 消息隊列的監控
高俊峰（南非螞蟻）
在企業實際應用中，如果業務比較復雜，那么管理的 Consumer Group 和 Topic 數也會隨之增加，此時如果再使用 Kafka 提供的命令行工具，可能會出現力不從心的感覺。
因此，我們需要一款更加智能的 Kafka 監控系統，目前企業使用比較多的有 Kafka Manager、Kafka Eagle 等。
Kafka Eagle 介紹
由於 Kafka Eagle 比 Kafka Manager 更簡單、更好用，所以這一課時將重點介紹 Kafka Eagle 這款 Kafka 可視化管理工具。
Kafka Eagle 是一款用來管理 Kafka 集群的可視化工具，它可以支持管理多個 Kafka 集群，還可以管理 Kafka 的 Topic（查看、刪除、創建等），也可以對消費者狀態進行監控，並可實現消息阻塞告警、集群健康狀態檢測等功能。
Kafka Eagle 安裝
點擊這里下載 Kafka-Eagle 安裝包，我下載的是 kafka-eagle-bin-2.0.0.tar.gz 版本，下載完成后，將安裝包解壓到 172.16.213.151 主機上，即可完成安裝，操作如下：
[root@nnmaster ~]#tar zxvf kafka-eagle-bin-2.0.0.tar.gz
[root@nnmaster ~]#cd kafka-eagle-bin-2.0.0
[root@nnmaster kafka-eagle-bin-2.0.0]# tar zxvf kafka-eagle-web-2.0.0-bin.tar.gz -C /usr/local
[root@nnmaster kafka-eagle-bin-2.0.0]# mv /usr/local/kafka-eagle-web-2.0.0 /usr/local/kafka-eagle
[root@nnmaster kafka-eagle-bin-2.0.0]# chown -R kafka:kafka /usr/local/kafka-eagle
上面這個操作步驟是解壓 Kafka Eagle，然后安裝到 /usr/local 目錄下，最后，授權 kafka-eagle 目錄為 Kafka 用戶權限，后面的管理配置操作都通過 Kafka 用戶來完成。
Kafka Eagle 安裝完成后，還需要配置 Java 環境變量和設置 Kafka Eagle 目錄，將下面內容加入 /home/kafka/.bashrc 文件中：
export KE_HOME=/usr/local/kafka-eagle
export PATH=$PATH:$KE_HOME/bin
export JAVA_HOME=/opt/bigdata/jdk
配置完成，執行 source /home/kafka/.bashrc 命令，使配置生效。
最后一個步驟，需要安裝一個數據庫，因為 Kafka Eagle 需要數據庫的支持，這里我采用 MySQL 數據庫，版本為 mysql5.7.30，不過安裝過程這里不再介紹。安裝完成后，需要做如下配置和授權：
mysql> create database ke character set utf8;
mysql> create user 'root'@'172.16.213.151' identified by 'xxxxxx';
mysql> grant all privileges on ke.* to 'root'@'172.16.213.151';
mysql> flush privileges;
在這個步驟中，首先創建了一個 ke 庫，是給 Kafka Eagle 使用的，然后創建了一個 root 用戶，並給 root 用戶授權給管理 ke 庫，最后刷新配置生效。
配置 Kafka Eagle 監控 Kafka 集群
Kafka Eagle 的配置文件位於 $KE_HOME/conf/ 目錄下，配置文件為 system-config.properties，配置完成的文件內容如下：
kafka.eagle.zk.cluster.alias=cluster1
cluster1.zk.list=172.16.213.152:2181,172.16.213.138:2181,172.16.213.80:2181
kafka.zk.limit.size=25
kafka.eagle.webui.port=8048
cluster1.kafka.eagle.offset.storage=kafka
kafka.eagle.metrics.charts=true
kafka.eagle.metrics.retain=15
kafka.eagle.sql.topic.records.max=5000
kafka.eagle.sql.fix.error=true
kafka.eagle.topic.token=keadmin
kafka.eagle.driver=com.mysql.jdbc.Driver
kafka.eagle.url=jdbc:mysql://172.16.213.151:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password=123456
其中，每個配置項含義如下。
-
kafka.eagle.zk.cluster.alias：用來指定需要配置的 Kafka 集群名稱，可以配置多個，用逗號分隔。
-
cluster1.zk.list：用來配置多個 Kafka 集群所對應的 ZooKeeper 集群列表。注意這個寫法，這是配置 cluster1 集群，如果有多個集群，依次填寫每個 Kafka 集群對應的 ZooKeeper 集群列表即可。
-
kafka.zk.limit.size：設置 ZooKeeper 客戶端最大連接數。
-
kafka.eagle.webui.port：設置 Kafka Eagle 的 Web 訪問端口，默認是 8048。
-
cluster1.kafka.eagle.offset.storage：設置存儲消費信息的類型，一般在 Kafka0.9 版本之前，消費信息會默認存儲在 ZooKeeper 中，所以存儲類型設置 zookeeper 即可。如果你使用的是 Kafka0.10 之后的版本，那么，消費者信息默認存儲在 Kafka 中，所以存儲類型需要設置為 kafka。同時，在使用消費者 API 時，盡量保證客戶端 Kafka API 版本和 Kafka 服務端的版本一致。
-
kafka.eagle.metrics.charts：設置是否開啟 Kafka Eagle 的監控趨勢圖，默認是不啟用方式，如果需要查看 Kafka 監控趨勢圖，則需要設置為 true。
-
kafka.eagle.metrics.retain：設置數據默認保留時間，這里的 15 表示 15 天。
-
kafka.eagle.sql.fix.error：在使用 KSQL 查詢 Topic 時，如果遇到錯誤，可以開啟這個屬性，默認不開啟。
-
kafka.eagle.sql.topic.records.max：KSQL 查詢 Topic 數據默認是最新的 5000 條，如果在使用 KSQL 查詢的過程中出現異常，可以將 kafka.eagle.sql.fix.error 的值設置為 true，Kafka Eagle 會在系統中自動修復錯誤。1
-
kafka.eagle.topic.token：設置在刪除 Kafka Topic 時的 Token 令牌，需要記住這個值。
-
kafka.eagle.driver：設置連接數據庫的驅動信息。
-
kafka.eagle.url：設置 jdbc 連接 MySQL 數據庫的地址。
-
kafka.eagle.username：設置連接到 MySQL 數據庫的用戶名。
-
kafka.eagle.password：設置連接到 MySQL 數據庫的用戶密碼。
上面配置中，我開啟了 Kafka Eagle 的監控趨勢圖，因此，還需要開啟 Kafka 系統的 JMX 端口，默認 Kafka 沒有開啟 JMX 端口，可修改 Kafka 啟動腳本 kafka-server-start.sh，添加如下內容：
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
export JMX_PORT="9999"
fi
注意，這個 9999 端口可修改為任意其他端口。在 Kafka 集群中所有節點添加這個配置，添加完成后，重啟所有 Kafka 集群節點以使配置生效。
Kafka Eagle 使用
1.啟動 Kafka Eagle 服務
所有配置完成后，就可以啟動 Kafka Eagle 服務了，執行如下命令啟動服務：
[kafka@nnmaster kafka-eagle]$ cd /usr/local/kafka-eagle/bin
[kafka@nnmaster bin]$ ./ke.sh start
[kafka@nnmaster bin]$ jps
7557 Jps
15855 KafkaEagle
服務啟動成功后，可以看到有個 KafkaEagle 進程標識，表示啟動成功，要查看 Kafka Eagle 啟動日志，可訪問 Kafka Eagle 的 logs 目錄，主要查看 log.log、error.log 及 ke_console.out 三個文件。如果啟動失敗，文件中會有失敗信息及失敗原因。
要關閉 Kafka Eagle 服務，執行“./ke.sh stop”即可。
Kafka Eagle 服務啟動后，即可打開 Web 界面，默認 Web 登錄用戶名為 admin，密碼為 123456，登錄端口為 8048，訪問 http://172.16.213.151:8048 地址即可打開登錄界面。
2.Kafka Eagle 的使用
Kafka Eagle 的使用比較簡單，下面是一個功能預覽界面圖：
在上圖中，對於左側導航欄，我已經將每個欄目的功能做了標注，整體分為 5 個部分，分別是全局視圖、Kakfa 消息管理、Kafka 集群監控、Kafka 監控告警及系統設置。
點擊上圖導航欄中的 BScreen 鏈接，即可進入下圖界面：
此界面是個 Kafka 實時狀態監控圖，主要展示了消息生產的狀態和消費狀態，數據實時更新，非常炫酷。
接着前面的功能預覽界面圖，在此圖右側部分，顯示了 Kafka 集群 Brokers 的數量、Topic 的數量、ZooKeeper 集群的節點數及消費者數量，點擊 Brokers 的數量鏈接，顯示下圖界面：
此界面顯示了 Kafka 集群的狀態信息，主要是 Kakfa 占用的系統 CPU、內存等資源信息，要獲得這些信息，需要在 Kafka 上開啟 JMX 端口，這里設置的 JMX 端口為 9999。
接着，回到功能預覽界面圖，再點擊 Topic 的鏈接，來到下圖所示的界面：
此界面顯示了 Kafka 集群中所有的 Topic 信息，包含 partitions 數量、數據偏差等，點擊任意一個 Topic 鏈接，來到下圖所示的界面：
此界面主要顯示了某個 Topic 中目前的消息數、容量以及每個 Partition 中存儲的消息數。
最后，再點擊功能預覽界面圖中的消費者鏈接，來到下圖所示的界面：
此界面主要顯示了目前有哪些消費組，可以看到這里有一個 Logstash 消費者組，點擊這個消費者組，然后點擊 Runing 按鈕，來到如下的界面：
此界面主要顯示了目前消費者消費 Nginxlogs 這個 Topic 的狀態信息，重點關注 Lag 列的信息，此信息表示消費延時，如果延時值過大，則表示消費速度過慢，需要引起注意。
下面再來看下 Kafka Eagle 提供的 KSQL 功能，如下圖所示：
此界面中，右邊的 select 語句，nginxlogs 表示 topic 的名稱，“partition in (0)”表示查詢哪個 partition 中的數據，最后的“limit 5”表示顯示前 5 條消息。
接着再來看下 Kafka 的監控指標，如下圖所示：
上圖曲線顯示了每分鍾推送到 Kafka 的消息數，另外還可以看到有很多其他監控指標，比如每秒生產多少數據量、每秒消費多少數據量等。
最后，再來看下如何配置監控告警，首先在告警欄中選擇創建一個告警組，如下圖所示：
此界面設置告警組名稱、告警方式和告警接口，支持釘釘、微信和郵件告警，推薦釘釘告警，配置比較簡單，直接給出一個告警接口地址即可。
接着，配置一個消費者延時告警，如下圖所示：
此界面中主要配置的是消費延時值，上面配置為 1，也就是說延時值超過 1 就進行告警，這個值要根據實際情況進行設置。在海量數據環境下（上億條數據），有幾千條消息延時，其實也是正常的，而在小量數據環境下（幾十萬條數據），延時消息超過百條，可能就是消費延時了，需要關注延時的原因。
最后，再來看下如何設置集群的告警，如下圖所示：
此界面中，配置的是對 Kafka 集群的告警，也可以設置 ZooKeeper、Topic、Producer 等的告警，告警可以設置告警級別、告警次數及告警組等信息。
所有配置完成后，就可以實現故障告警了，下圖是一個釘釘告警截圖：
從圖中可以看出，此告警是消費延時導致的，我們設置的最大消費延時為 1，而目前延時是 793，這已經是第三次告警了，后面將不再進行告警，因為我配置的最大告警次數為 3。
總結
本課時主要講解了 Kafka Eagle 的安裝、配置與使用，其是目前最流行的 Kafka 監控利器，很多大企業都在使用它來監控和管理 Kafka，快速、熟練使用 Kafka Eagle 是本課程的重點。