如何通过Spring Kafka获取指定Kafka Topic的消费者延迟？

问题分析 #

你之前尝试的metrics方式无法满足需求，因为records-lag指标通常按消费者实例或分区聚合，没有直接关联到具体Topic，没法精准过滤出单个Topic的未消费消息数。

解决方案 #

计算指定Topic的consumer lag，核心是对比消费者组已提交的偏移量和Topic分区的最新末端偏移量，累加对应分区的差值即可。

1. 配置AdminClient #

首先配置Kafka AdminClient，用于获取消费者组和Topic的元数据：

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public AdminClient adminClient() {
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        return AdminClient.create(config);
    }
}

2. 实现lag计算方法 #

通过AdminClient获取所需数据，实现getUnconsumedMessagesCount方法：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@Component
public class KafkaLagService {

    private final AdminClient adminClient;
    private final KafkaListenerEndpointRegistry endpointRegistry;

    public KafkaLagService(AdminClient adminClient, KafkaListenerEndpointRegistry endpointRegistry) {
        this.adminClient = adminClient;
        this.endpointRegistry = endpointRegistry;
    }

    public int getUnconsumedMessagesCount(String topicName) throws ExecutionException, InterruptedException {
        // 从监听容器获取消费者组ID
        String groupId = endpointRegistry.getListenerContainer("errorTopicsConsumer")
                .getGroupId();

        // 获取消费者组关联的分区（消费者已启动时）
        ConsumerGroupDescription groupDesc = adminClient.describeConsumerGroups(Collection.of(groupId))
                .describedGroups()
                .get(groupId)
                .get();

        Collection<TopicPartition> topicPartitions = groupDesc.members().stream()
                .flatMap(member -> member.assignment().topicPartitions().stream())
                .filter(tp -> tp.topic().equals(topicName))
                .collect(Collectors.toList());

        // 消费者未启动时，直接获取Topic的所有分区
        if (topicPartitions.isEmpty()) {
            TopicDescription topicDesc = adminClient.describeTopics(Collection.of(topicName))
                    .all()
                    .get()
                    .get(topicName);
            topicPartitions = topicDesc.partitions().stream()
                    .map(partitionInfo -> new TopicPartition(topicName, partitionInfo.partition()))
                    .collect(Collectors.toList());
        }

        // 获取消费者组已提交的偏移量
        ListConsumerGroupOffsetsResult offsetsResult = adminClient.listConsumerGroupOffsets(groupId);
        Map<TopicPartition, Optional<Long>> committedOffsets = offsetsResult.partitionsToOffsetAndMetadata()
                .get();

        // 获取分区最新末端偏移量
        Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(topicPartitions.stream()
                        .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())))
                .all()
                .get();

        // 累加所有分区的lag
        int totalLag = 0;
        for (TopicPartition tp : topicPartitions) {
            long committedOffset = committedOffsets.getOrDefault(tp, Optional.empty()).orElse(0L);
            long endOffset = endOffsets.getOrDefault(tp, 0L);
            totalLag += Math.max(0, endOffset - committedOffset);
        }

        return totalLag;
    }
}

3. 关键说明 #

• 若消费者未启动（autoStartup=false），消费者组无活跃成员，此时直接获取Topic所有分区，用组内最新提交的偏移量计算lag。
• 需处理异步操作的异常（ExecutionException、InterruptedException），实际业务中可根据场景封装或转换。
• AdminClient配置需与消费者配置保持一致，尤其是bootstrap-servers。

内容的提问来源于stack exchange，提问作者Aryak Deshpande