目標

獲取某 topic 共生產(chǎn)多少條消息。

創(chuàng)建測試 topic

bin/kafka-topics.sh -zookeeper localhost:2181 --create --partitions 5 --replication-factor 1 --topic test765

生產(chǎn)測試消息

為 topic test765 生產(chǎn) 500000 條消息：

bin/kafka-producer-perf-test.sh  --topic test765 --throughput  -1  --record-size 10 --num-records 500000 --producer-props bootstrap.servers=localhost:9092

獲取 topic 消息數(shù)

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test765 --time -1

# 輸出信息
test765:0:100000
test765:1:100000
test765:2:100000
test765:3:100000
test765:4:100000

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test765 --time -2

# 輸出信息
test765:0:0
test765:1:0
test765:2:0
test765:3:0
test765:4:0

--time-1 表示要獲取指定topic所有分區(qū)當前的最大位移，--time-2 表示獲取當前最早位移。

兩個命令的輸出結果相減便可得到所有分區(qū)當前的消息總數(shù)。

分區(qū)當前的消息總數(shù) = [--time-1] - [--time-2]

相減是因為隨著 kafka 的運行，topic 中有的消息可能會被刪除，，因此 --time-1 的結果其實表示的是歷史上該topic生產(chǎn)的最大消息數(shù)，如果用戶要統(tǒng)計當前的消息總數(shù)就必須減去 --time-2 的結果。

本例中沒有任何消息被刪除，故 --time-2 的結果全是0，表示最早位移都是0，消息總數(shù)等于歷史上發(fā)送的消息總數(shù)。