Kafka 主题中的消息数答案

【问题标题】：Kafka Number of messages in a topicKafka 主题中的消息数
【发布时间】：2020-07-31 20:09:31
【问题描述】：

我需要存储 kafka 主题中的消息数。这与是否有消费者消费了消息无关。

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092,localhost:9093,localhost:9094 --topic test-topic

上面给出了主题的偏移量？

以上是否等于kafka topic当前存储的消息数？

【问题讨论】：

标签： apache-kafka

【解决方案1】：

不完全是。您获得的数字仅指所有主题分区的当前最大偏移量。消息计数还取决于该主题的分区开始偏移量。

你可以跑

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092,localhost:9093,localhost:9094 --topic test-topic --time -1

和

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092,localhost:9093,localhost:9094 --topic test-topic --time -2

分别通过 endOffsets 减去 beginOffsets 来计算每个分区的消息数，然后将它们相加得到该主题的总记录数。

【讨论】：

这不适用于压缩主题
当我压缩时，我们会丢失任何数据。在没有任何数据丢失的情况下交叉验证压缩成功应用的正确方法是什么？

【解决方案2】：

是的，如果最早的偏移量等于零，这等于消息数。如果最早的偏移量不等于零，则需要计算差值，然后每个分区求和。

【讨论】：

这不适用于压缩主题

【解决方案3】：

上面给出了主题的偏移量？ 是的，它给出了当前的最大偏移量

上面是否等于kafka topic当前存储的消息数？ 不，这不是 kafka 中的消息数，因为在保留期之后，消息将从主题中删除，因此偏移量！= 消息数

获取kafka中的消息数量

    brokers="<broker1:port>"
topic=<topic-name>
sum_1=$(/usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list $brokers --topic $topic --time -1 | grep -e ':[[:digit:]]*:' | awk -F  ":" '{sum += $3} END {print sum}')
sum_2=$(/usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list $brokers --topic $topic --time -2 | grep -e ':[[:digit:]]*:' | awk -F  ":" '{sum += $3} END {print sum}')
echo "Number of records in topic ${topic}: "$((sum_1 - sum_2))

其中选项 --time -1 => 当前最大偏移量 & --time -2 是当前最小偏移量

【讨论】：