基本相同，但必须检查zkutils.consumerspath（=/consumers）。
zookeeper中的consumers结构是下一个/consumers/[groupid]/ids/[consumerid]，因此导航时可以获得每个组的组和使用者。

对于0.9.x新消费者，请列出所有活动消费者组：
找到所有经纪人，向每个经纪人发送“listgroups”请求，获取所有的组信息；
有关详细信息，请参阅 $KAFKA_HOME/bin/kafka-consumer-groups.sh （kafka.admin.consumergroupcommand.kafkaconsumergroupservice.list（））
对于0.9.x新消费者，请描述特定消费者群体的详细信息：
找到用户组协调器，向其发送“describegroups”请求，获取所有组成员信息和分区分配信息；
调用kafkaconsumer.committed（topicpartition partition）获取给定分区的最后一个提交的偏移量。
有关详细信息，请参阅 $KAFKA_HOME/bin/kafka-consumer-groups.sh （kafka.admin.consumergroupcommand.kafkaconsumergroupservice.describe（））
请注意，旧使用者和新使用者对此有完全不同的实现（这两种逻辑都在kafka.admin.consumergroupcommand中实现）。

我不确定你到底需要什么信息，但我做了一个示例程序，它提供了与kafka-consumer-groups.sh相同的信息——描述。
要使用此代码，请将此依赖项添加到pom中。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.9.0.1</version>
</dependency>

然后：

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import kafka.admin.AdminClient;
import kafka.coordinator.GroupOverview;

Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
AdminClient adminClient = AdminClient.create(props);

List<GroupOverview> groups =  scala.collection.JavaConversions.seqAsJavaList(
        adminClient.listAllConsumerGroupsFlattened());
for (GroupOverview group : groups) {
    String groupId = group.groupId();

    Properties consProps = new Properties();
    consProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
    consProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    consProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    consProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
    consProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    consProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer consumer = new KafkaConsumer(consProps);

    List<AdminClient.ConsumerSummary> groupSummaries = scala.collection.JavaConversions.seqAsJavaList(
            adminClient.describeConsumerGroup(groupId));

    System.out.println("GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER");

    for (AdminClient.ConsumerSummary summary : groupSummaries) {
        String owner = summary.clientId() + "_" + summary.clientHost();
        List<TopicPartition> topicPartitions = scala.collection.JavaConversions.seqAsJavaList(
                summary.assignment());
        for (TopicPartition tp : topicPartitions) {

            // Get current offset
            long currentOffset = consumer.committed(tp).offset();

            // get log end offset
            consumer.assign(Arrays.asList(tp));
            consumer.seekToEnd();
            long logEndOffset = consumer.position(tp);

            long lag = logEndOffset - currentOffset;

            System.out.println(groupId + ", " + tp.topic() + ", " + tp.partition() + ", " +
                    currentOffset + ", " + logEndOffset + ", " + lag + ", " + owner);
        }
    }
}