在前面的文章《Kafka的Lag计算误区及正确实现》中介绍了如何计算消费者的消费滞后量(Lag)，并且讲解了如何调用Kafka的kafka.admin.ConsumerGroupCommand文件中的KafkaConsumerGroupService来发送OffsetRequest和OffsetFetchRequest两个请求，进而通过两个请求结果之间的差值来获得结果。不过如果你不想修改kafka-core的代码并重新编译的话，这种实现方式无法成功，所以本文的主要目的就是通过调用更底层的API来实现不修改kafka-core的代码来实现KafkaConsumerGroupService的功能，即通过Java调用Scala的代码来实现获取Kafka的消费者详情的功能。
成都创新互联公司主营洛扎网站建设的网络公司,主营网站建设方案,成都APP应用开发,洛扎h5小程序开发搭建,洛扎网站营销推广欢迎洛扎等地区企业咨询实现如同 bin/kafka-consumer-group.sh –describe –bootstrap-server localhost:9092 –group CONSUMER_GROUP_ID的效果：
[root@node2 kafka_2.12-1.0.0]# bin/kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group CONSUMER_GROUP_ID
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
topic-test1 0 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID
topic-test1 1 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID
topic-test1 2 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID
topic-test1 3 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_IDKafkaConsumerGroupService的核心方法是CollectGroupAssignment，其方法参数为一个consumer group的groupId，方法输出为上面示例中的列表信息。CollectGroupAssignment方法主要有以下几个步骤：
对应Java版的KafkaConsumerGroupService改造代码可以参见代码，目录结构如下图所示：
其中model中的ConsumerGroupSummary、ConsumerSummary和PartitionAssignmentState是简单的JavaBean, PartitionAssignmentState是用来保存每个TopicPartition的消费者信息的，具体内容参考如下。KafkaConsumerGroupCustomService就是本文所要陈述的Java改造办的KafkaConsumerGroupSerivice，ConsumerGroupUtils用来存放一些公用的代码。
@Data
@Builder
public class PartitionAssignmentState {
private String group; // groupId
private Node coordinator; // consumer coodinator节点信息
private String topic;
private int partition;
private long offset;
private long lag;
private String consumerId;
private String host;
private String clientId;
private long logEndOffset;
}初始化KafkaConsumerGroupCustomService需要Kafka的服务端地址，然后初始化AdminClient和KafkaConsumer，AdminClient中包含了众多管理类方法，主要是通过发送各种自定义协议请求来完成，上面步骤中所说的describeConsumerGroup和listGroupOffsets方法也是通过AdminClient来实现的；KafkaConsumer主要是用来获取TopicPartition对应的HW(消费者可见的LogEndOffsets)的。
KafkaConsumerGroupCustomService中与scala版对应的collectGroupAssignment方法如下（详细步骤参考代码注释）：
public List collectGroupAssignment(
AdminClient adminClient, KafkaConsumer consumer,
String group) {
//1. 获取consumer group的基本信息，包括CONSUMER-ID、HOST、
// CLIENT-ID以及TopicPartition信息
AdminClient.ConsumerGroupSummary consumerGroupSummary
= adminClient.describeConsumerGroup(group, 0);
List assignedTopicPartitions = new ArrayList<>();
List rowsWithConsumer = new ArrayList<>();
scala.collection.immutable.List consumers
= consumerGroupSummary.consumers().get();
if (consumers != null) {
//2. 获取各个分区(Partition)的对应的消费位移CURRENT-OFFSET
scala.collection.immutable.Map offsets
= adminClient.listGroupOffsets(group);
if (offsets.nonEmpty()) {
String state = consumerGroupSummary.state();
// 3. 还有一个状态是Dead表示"group"对应的consumer group不存在
if (state.equals("Stable") || state.equals("Empty")
|| state.equals("PreparingRebalance")
|| state.equals("AwaitingSync")) {
List consumerList = changeToJavaList(consumers);
// 4. 获取当前有消费者的消费信息，即包含CONSUMER-ID、HOST、CLIENT-ID
rowsWithConsumer = getRowsWithConsumer(consumerGroupSummary, offsets,
consumer, consumerList, assignedTopicPartitions, group);
}
}
//5. 获取当前没有消费者的消费信息
List rowsWithoutConsumer =
getRowsWithoutConsumer(consumerGroupSummary,
offsets, consumer, assignedTopicPartitions, group);
//6. 合并结果
rowsWithConsumer.addAll(rowsWithoutConsumer);
}
return rowsWithConsumer;
} KafkaConsumerGroupCustomService类中包含有getRowsWithConsumer()、getRowsWithoutConsumer()、changeToJavaList等私有方法也都是在Scala语言与Java语言之间进行切换，这样可以不需要修改kafka-core的原生代码而通过外部的封装调用既可以实现获取Kafka消费者详情的功能。光看代码比较抽象，建议对此感兴趣的同学可以亲自对比一下kafka-core包中kafka.admin.ConsumerGroupCommand的KafkaConsumerGroupSerivice与笔者自定义的KafkaConsumerGroupCustomService的实现来了解下Scala语言到Java语言的转换。
如果需要打印详情可以调用KafkaConsumerGroupCustomService同目录的ConsumerGroupUtils类中的printPasList(List list)方法。注意要运行这些代码需要JDK8的环境，笔者为了让代码显得“骚气”一点就用来一点Java8的语法，如果需要Java7的代码实现可以关注私聊。
或许有些同学对于Scala和Java交叉的代码并不感冒，想要寻求一种存Java式的实现方式，那么在这里怎么实现呢？答案是通过KafkaAdminClient，它是AdminClient的Java版实现，从Kafka0.11.0.0版本开始引入的，不过KafkaAdminClient本身并没有提供describeConsumerGroup、listGroupOffsets之类的方法给我们直接使用，扩展一下也很方便，由于篇幅限制，这部分的内容将在下一篇文章中进行介绍，如果想要先一睹为快，可以参考下代码实现，详细的逻辑解析敬请期待….
本文的重点是你有没有收获与成长，其余的都不重要，希望读者们能谨记这一点。同时我经过多年的收藏目前也算收集到了一套完整的学习资料，包括但不限于：分布式架构、高可扩展、高性能、高并发、Jvm性能调优、Spring，MyBatis，Nginx源码分析，Redis，ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多个知识点高级进阶干货，希望对想成为架构师的朋友有一定的参考和帮助
另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
