计算 kafka 主题中存储的消息数

Question

我使用的是0.9.0.0版本的Kafka，我想在不使用管理脚本Kafka-console-consumer.sh的情况下计算主题中的消息数。

我已经尝试了答案Java“How to get number of messages in a topic in apache kafka”中的所有命令，但都没有结果。有人能帮我吗？

Answer 1

您可以使用以下命令对所有计数求和:

.../bin/kafka-run-class kafka.tools.GetOffsetShell --broker-list <<broker_1>>:9092,<<broker_2:9092>>... --topic <<your_topic_name>> --time -1 | while IFS=: read topic_name partition_id number; do echo "$number"; done | paste -sd+ - | bc

Answer 2

从技术上讲，您可以简单地使用主题中的所有消息并计算它们：

例子：

kafka-run-class.sh kafka.tools.SimpleConsumerShell --broker-list localhost:9092 --topic XYZ --partition 0*

但是kafka.tools.GetOffsetShell方法会给你偏移量，而不是主题中实际的消息数量。这意味着如果主题被压缩，如果你通过消费消息或读取偏移量来计算消息，你会得到两个不同的数字。

主题压缩：https://kafka.apache.org/documentation.html#design_compactionbasics

Answer 3

您可以尝试执行以下命令:

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

然后，汇总每个分区的所有计数。

更新:Java实现

Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
......
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList("your_topic"));
    Set<TopicPartition> assignment;
    while ((assignment = consumer.assignment()).isEmpty()) {
        consumer.poll(Duration.ofMillis(100));
    }
    final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignment);
    final Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(assignment);
    assert (endOffsets.size() == beginningOffsets.size());
    assert (endOffsets.keySet().equals(beginningOffsets.keySet()));

    Long totalCount = beginningOffsets.entrySet().stream().mapToLong(entry -> {
            TopicPartition tp = entry.getKey();
            Long beginningOffset = entry.getValue();
            Long endOffset = endOffsets.get(tp);
            return endOffset - beginningOffset;
        }).sum();
    System.out.println(totalCount);
}