查看kafka消息消費狀況

查看主題命令

#展現topic列表 ./kafka-topics.sh --list --zookeeper 172.18.153.12:2188

#描述topic ./kafka-topics.sh --describe --zookeeper 172.18.153.12:2188 --topic test

#查看topic某分區偏移量最大（小）值 ./kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --broker-list 10.1.3.84:9098 --partitions 0

#增長topic分區數 ./kafka-topics.sh --zookeeper 172.18.153.12:2188 --alter --topic test --partitions 10

#刪除topic：慎用，只會刪除zookeeper中的元數據，消息文件須手動刪除 方法一： ./kafka-topics.sh --delete --zookeeper 172.18.153.12:2188 --topic test 方法二：待驗證 ./kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper 172.18.153.12:2188 --topic test

#查看topic消費進度，必須參數爲--group， 不指定--topic，默認爲全部topic, ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group group1

#列出全部主題中的全部用戶組： ./kafka-consumer-groups.sh --bootstrap-server 10.1.3.84:9098 --list

#要使用ConsumerOffsetChecker查看上一個示例中消費者組的偏移量，咱們按以下所示「describe」消費者組： ./kafka-consumer-groups.sh --bootstrap-server 10.1.3.84:9098 --describe --group group1

#-members: 此選項提供使用者組中全部活動成員的列表。 ./kafka-consumer-groups.sh --bootstrap-server 10.1.3.84:9098 --describe --group group1 --members

查看kafka消息消費狀況

消息堆積是消費滯後(Lag)的一種表現形式，消息中間件服務端中所留存的消息與消費掉的消息之間的差值即爲消息堆積量，也稱之爲消費滯後(Lag)量。 對於Kafka而言，消息被髮送至Topic中，而Topic又分紅了多個分區(Partition)，每個Partition都有一個預寫式的日誌文件，雖然Partition能夠繼續細分爲若干個段文件(Segment)，可是對於上層應用來講能夠將Partition當作最小的存儲單元(一個由多個Segment文件拼接的「巨型文件」)。 每一個Partition都由一系列有序的、不可變的消息組成，這些消息被連續的追加到Partition中。咱們來看下圖，其就是Partition的一個真實寫照：

上圖中有四個概念：

LogStartOffset：表示一個Partition的起始位移，初始爲0，雖然消息的增長以及日誌清除策略的影響，這個值會階段性的增大。 ConsumerOffset：消費位移，表示Partition的某個消費者消費到的位移位置。 HighWatermark：簡稱HW，表明消費端所能「觀察」到的Partition的最高日誌位移，HW大於等於ConsumerOffset的值。 LogEndOffset：簡稱LEO, 表明Partition的最高日誌位移，其值對消費者不可見。

好比在ISR（In-Sync-Replicas）副本數等於3的狀況下（以下圖所示），消息發送到Leader A以後會更新LEO的值，Follower B和Follower C也會實時拉取Leader A中的消息來更新本身，HW就表示A、B、C三者同時達到的日誌位移，也就是A、B、C三者中LEO最小的那個值。因爲B、C拉取A消息之間延時問題，因此HW必然不會一直與Leader的LEO相等，即LEO>=HW。

要計算Kafka中某個消費者的滯後量很簡單，首先看看其消費了幾個Topic，而後針對每一個Topic來計算其中每一個Partition的Lag，每一個Partition的Lag計算就顯得很是的簡單了，參考下圖：

由圖可知消費Lag=HW - ConsumerOffset。Kafka中自帶的kafka-consumer_groups.sh腳本中就有Lag的信息，示例以下：

[root@node2 kafka_2.12-1.0.0]# bin/kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group CONSUMER_GROUP_ID

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID topic-test1 0 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID topic-test1 1 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID topic-test1 2 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID topic-test1 3 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID

參考： Kafka的Lag計算誤區及正確實現：https://blog.csdn.net/u013256816/article/details/79955578 如何使用JMX監控Kafka：http://www.javashuo.com/article/p-hnrvttdm-mb.html