kafka消费者列表API答案

【问题标题】：kafka consumer list APIkafka消费者列表API
【发布时间】：2018-05-25 10:39:17
【问题描述】：

我是卡夫卡的新手。是否有任何 API 可以列出与 kafka-consumer-groups.sh/bat 文件相同的结果。

kafka-consumer-groups --bootstrap-server localhost:9091 --list

【问题讨论】：

是...这些shell命令都运行Java类

【解决方案1】：

感谢您的帮助。这是下面的代码来获取消费者列表的值

public static void main(String[] args) {
String[] arg = new String[3];
arg[0] = "--bootstrap-server";
arg[1] = " localhost:9092";
arg[2] = "--list";

String[] resultArray = new String[3];
ConsumerGroupCommand.ConsumerGroupCommandOptions checkArgs = new ConsumerGroupCommand.ConsumerGroupCommandOptions(
    arg);
checkArgs.checkArgs();
List<String> result = new ConsumerGroupCommand.KafkaConsumerGroupService(checkArgs).listGroups();
System.out.println(result);
result.copyToArray(resultArray);
java.util.List<String> names = java.util.Arrays.asList(resultArray);

System.out.println("result" + names);

//the below line will print the value in console 
ConsumerGroupCommand.main(arg);
 }

【讨论】：

【解决方案2】：

最好的办法是使用 KafkaAdminClient 类（如果您使用的是 0.11 或更高版本（参见 https://cwiki.apache.org/confluence/display/KAFKA/KIP-117%3A+Add+a+public+AdminClient+API+for+Kafka+admin+operations）

ConsumerGroupCommand 添加到 KafkaAdminClient 是 AK 2.0 版本：https://cwiki.apache.org/confluence/display/KAFKA/KIP-222+-+Add+Consumer+Group+operations+to+Admin+API

对于旧版本，正如@cricket_007 所指出的，您可以查看脚本以查看它实际使用的类，您可以将此类嵌入到您自己的代码中以直接调用它。

【讨论】：

感谢@Matthias J. Sax 和@cricket_007 的快速回复。检查管理客户端 API 后，我找不到任何符合我要求的 API。然后我开始寻找 Kafka-consumer-groups.bat 文件中使用的 java 类，发现下面的命令可能正在运行 ConsumerGroupCommand。这是用Scala编写的。有什么简单的方法可以处理吗？
它正在运行的命令 :: "java" -Xmx256M -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC -Djava.awt.headless =true -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=confluent-3.2.0\confluent-3.2 .0/logs -cp .;\share\java\kafka\kafka-clients-0.10.2.0-cp1.jar";"\share\java\kafka*" kafka.admin.ConsumerGroupCommand --bootstrap-server localhost:9092 --list
刚刚意识到ConsumerGroupCommand 在即将发布的 2.0 版本中添加到 KafkaAdminClient。如果您等不及了，您可以向<artifactId>kafka-core</artifactId> 添加一个依赖项，然后通过调用ConsumerGroupCommand#main() 并传入相应的命令行参数，直接在您自己的代码中使用kafka.admin.ConsumerGroupCommand。或者，作为（更好的）替代方案，挖掘ConsumerGroupCommand 并提取发出实际请求的代码。在内部，ConsumerGroupCommand 使用旧的 kafka.admin.AdminClient 向代理发出实际请求。
@MatthiasJ.Sax 如果我错了，请纠正我。 AdminClient 较新的消费者组方法仅提供每个消费者组的已提交偏移量。您知道是否有办法获取 AdminClient 中每个主题的滞后和日志结束偏移量？如果不是，我将如何在 java 中获得类似的东西？
“每个主题”是什么意思——提交的偏移量与消费者群体和主题相关联。因此，没有全局“滞后”的概念，而“滞后”仅针对消费者组定义。您可以获取所有主题的列表，也可以使用KafkaConsumer 获取所有这些主题的所有分区的endOffset()。

【解决方案3】：

public List<String> listConsumersGroups() throws ExecutionException, InterruptedException {
    AdminClient admin = AdminClient.create(config);
    ListConsumerGroupsResult result = admin.listConsumerGroups();
    Collection<ConsumerGroupListing> listings = result.valid().get();
    List<String> consumerGroups = new ArrayList<>(listings.size());
    listings.forEach(l->consumerGroups.add(l.groupId()));
    return consumerGroups; 
}

【讨论】：