如何通过Java代码获取Kafka指定topic的消费者数量

实现思路 #

你可以直接使用Kafka官方提供的AdminClient API完成该检测，无需调用外部shell脚本，完全适配生产环境使用，核心逻辑是查询集群所有活跃消费者组的订阅信息和成员状态，筛选出订阅了目标Topic的活跃消费者。

前置依赖 #

Maven项目引入kafka-clients依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>和你的Kafka集群版本保持一致</version>
</dependency>

核心实现代码 #

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.ConsumerGroupState;
import java.util.*;
import java.util.concurrent.ExecutionException;

public class KafkaActiveConsumerChecker {
    // 传入Kafka集群地址和目标Topic名称，返回是否存在其他活跃消费者
    public boolean hasActiveConsumers(String bootstrapServers, String targetTopic) throws ExecutionException, InterruptedException {
        // 配置AdminClient
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 生产环境如有SASL、SSL等认证配置，可在此处添加对应参数

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 1. 获取集群所有消费者组ID
            List<String> allGroupIds = adminClient.listConsumerGroups()
                    .all()
                    .get()
                    .stream()
                    .map(ConsumerGroupListing::groupId)
                    .toList();

            // 2. 批量查询所有消费者组的详细信息
            DescribeConsumerGroupsResult groupsResult = adminClient.describeConsumerGroups(allGroupIds);
            Map<String, ConsumerGroupDescription> groupDescMap = groupsResult.all().get();

            for (Map.Entry<String, ConsumerGroupDescription> entry : groupDescMap.entrySet()) {
                ConsumerGroupDescription groupDesc = entry.getValue();
                // 只筛选处于活跃状态的消费者组
                if (groupDesc.state() != ConsumerGroupState.STABLE && groupDesc.state() != ConsumerGroupState.PREPARING_REBALANCE) {
                    continue;
                }
                // 检查组内成员的订阅Topic是否包含目标Topic
                for (MemberDescription member : groupDesc.members()) {
                    Optional<MemberAssignment> assignmentOpt = member.assignment();
                    if (assignmentOpt.isPresent()) {
                        boolean isSubscribedTarget = assignmentOpt.get()
                                .topicPartitions()
                                .stream()
                                .anyMatch(tp -> tp.topic().equals(targetTopic));
                        if (isSubscribedTarget) {
                            // 存在订阅目标Topic的活跃消费者
                            return true;
                        }
                    }
                }
            }
        }
        // 无匹配的活跃消费者
        return false;
    }
}

注意事项 #

• 应用启动时调用该方法，根据返回结果判断是否启动本地消费者即可
• 如果你的消费者属于固定消费者组，可直接传入组ID查询，无需遍历全量消费者组，提升查询效率
• 生产环境使用建议添加AdminClient请求超时配置，避免请求阻塞影响启动流程

内容的提问来源于stack exchange，提问作者Arpan Parikh