在Java中获取Kafka未消费消息计数

Answer 1

还有另一种替代方法

Kafka消费者有API来获取主题每个分区的endpoint

List partitions = new ArrayList<>();
        for (PartitionInfo p : parts) {
            partitions.add(new TopicPartition(topic, p.partition()));
        }
Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);

对于每个主题分区，u可以得到最近提交的偏移量。通过使用这两个数字，你可以很容易地得到未消耗的滞后

for (TopicPartition tp : offsets.keySet()) {
            OffsetAndMetadata commitOffset = consumer.committed(new TopicPartition(tp.topic(), tp.partition()));
            Long lag = commitOffset == null ? offsets.get(tp) : offsets.get(tp) - commitOffset.offset();            
        }