使用Python confluent_kafka查询监听指定Topic的消费者consumer.id方法

1. 命令行工具查询（Confluent Platform自带） #

你已经获取到所有消费组及对应订阅Topic的前提下，对每个订阅了目标Topic的消费组执行如下命令即可拿到组内所有活跃消费者ID：

# 命令路径：confluent-<版本号>/bin/kafka-consumer-groups.sh
./kafka-consumer-groups.sh --bootstrap-server <Kafka集群bootstrap地址> --group <目标消费组ID> --describe

输出结果的CONSUMER-ID列即为你需要的消费者ID，可自行过滤TOPIC列匹配你的目标Topic，筛选出所有符合要求的消费者。
如果需要批量查询所有消费组，可通过Shell脚本循环遍历你已拿到的消费组列表，重复执行上述命令过滤即可。

2. Confluent Control Center 页面查询 #

如果你部署了Confluent官方管控平台，直接进入目标Topic的详情页，切换到「消费者」标签，即可直观看到所有订阅该Topic的消费组，展开消费组就能直接查看组内每个在线消费者的ID、客户端ID、部署主机等信息。

3. 客户端API查询（以Java API为例，其他语言逻辑一致） #

可以通过Kafka官方AdminClient接口直接编码查询，无需依赖外部命令：

• 调用listConsumerGroups()接口获取集群内所有消费组ID
• 遍历每个消费组，调用describeConsumerGroups(Collections.singleton(groupId))获取消费组的完整详情
• 从返回的ConsumerGroupDescription对象中取members()集合，每个成员的consumerId()字段即为消费者ID；同时检查成员assignment()中包含的TopicPartition是否属于你要查询的目标Topic，匹配成功即为目标消费者
  示例代码片段：

// 初始化AdminClient
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka集群地址:9092");
try (AdminClient admin = AdminClient.create(props)) {
    String targetTopic = "指定的目标Topic名称";
    // 遍历所有消费组
    for (ConsumerGroupListing groupListing : admin.listConsumerGroups().all().get()) {
        String groupId = groupListing.groupId();
        ConsumerGroupDescription groupDesc = admin.describeConsumerGroups(Collections.singleton(groupId)).all().get().get(groupId);
        // 遍历消费组内所有活跃成员
        for (MemberDescription member : groupDesc.members()) {
            // 校验该消费者是否订阅了目标Topic
            boolean isMatch = member.assignment().topicPartitions().stream()
                    .anyMatch(tp -> tp.topic().equals(targetTopic));
            if (isMatch) {
                System.out.println("匹配的消费者ID：" + member.consumerId());
            }
        }
    }
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

注意：以上所有方法只能查询到当前在线活跃的消费者，已经掉线的消费者不会出现在返回结果中，符合「正在监听」的查询要求。

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