阅读全文

⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 https://blog.csdn.net/u013256816/article/details/80032594 「朱小厮」欢迎转载，保留摘要，谢谢！

---

🙂🙂🙂关注**微信公众号：【芋道源码】**有福利：

1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
4. 新的源码解析文章实时收到通知。每周更新一篇左右。
5. 认真的源码交流微信群。

---

在《Kafka的Lag计算误区及正确实现》一文中提及了kafka.admin.ConsumerGroupCommand.PartitionAssignmentState无法被外部访问，故要么将PartitionAssignmentState前的protected修饰符去掉，要么像《 如何获取Kafka的消费者详情》和《集群管理工具KafkaAdminClient——改造》这两篇这样来实现，但是真的需要这样子做么？

可以直接将describeGroup返回的结果转换成JSON然后传至监控页面（supported by YANGliiN oba）。代码如下：

String[] agrs = {"--describe", "--bootstrap-server", brokers, "--group", groupId};
ConsumerGroupCommand.ConsumerGroupCommandOptions options =
        new ConsumerGroupCommand.ConsumerGroupCommandOptions(agrs);
ConsumerGroupCommand.KafkaConsumerGroupService kafkaConsumerGroupService =
        new ConsumerGroupCommand.KafkaConsumerGroupService(options);
ObjectMapper mapper = new ObjectMapper();
//1. 使用jackson-module-scala_2.12
mapper.registerModule(new DefaultScalaModule());
//2. 反序列化时忽略对象不存在的属性
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
//3. 将Scala对象序列化成JSON字符串
String source = mapper.writeValueAsString(kafkaConsumerGroupService.describeGroup()._2.get());

这里需要采用的是jackson-module-scala的包实现，如果直接用普通的JSON序列化方式那么会达不到想要的效果，jackson以及jackson-module-scala对应的Maven库如下：

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-core</artifactId>
    <version>2.9.4</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.module</groupId>
    <artifactId>jackson-module-scala_2.12</artifactId>
    <version>2.9.5</version>
</dependency>

注意如果本地安装的Scala版本与所配置的jackson-module-scala版本不一致的话会报出一些异常。发散一下思维：既然可以序列化为JSON，那么完全可以通过JSON再反序列化会对象，只不过通过JSON作为中间媒介，将访问受限的Scala对象转变为Java对象，上面剩余代码如下：

//4. 将JSON字符串反序列化成Java对象
List<PartitionAssignmentState> target = mapper.readValue(source,
        getCollectionType(mapper,List.class,PartitionAssignmentState.class));
//5. 排序
target.sort((o1, o2) -> o1.getPartition() - o2.getPartition());
//6. 打印
printPasList(target);

如此就可以达到与前面几篇文章中关于获取消费者详情功能同样的效果。这里有两个注意要点：

1. PartitionAssignmentState中的coordinator是Node类型，这个类型需要自定义，Kafka原生的会报错。
2. 反序列化时Node会有一个empty的属性不识别，解决方案参考代码中的步骤2.

代码更多细节请参考：https://github.com/hiddenzzh/kafka/blob/master/src/main/java/com/hidden/custom/kafka/admin/KafkaConsumerGroupScalaService.java

通过JSON的序列化和反序列化操作实现了原本不能为之的事情，那么思维再发散一下，也可以序列化成字节流，比如通过ByteBuffer进行转换，只不过编程逻辑变得复杂了。

上面这段陈述有可能会让人觉得Scala与Java之间的互操作起来不容易，其实不然，上面这段陈述只是用来补充一下如何获取消费者详情的另一种方法，Scala与Java之间的互操作还是比较简单的，一般情况下都可以直接使用对方的类。对于集合而言，Scala中还有用于Scala与Java集合的互转的scala.collection.JavaConverters（scala2.8.1开始引入），与此雷同的scala.collection.JavaConversions已被标注为@Deprecated（since 2.12.0）。在scala代码中如果需要集合转换，首先引入scala.collection.JavaConverters._，进而显示调用asJava或者asScala方法完成转型。关于Scala与Java集合互转的介绍会在下一篇文章中呈现。

666. 彩蛋

如果你对 Kafka 并发感兴趣，欢迎加入我的知识一起交流。