---

歡迎支持筆者新作：《深入理解Kafka:核心設計與實踐原理》和《RabbitMQ實戰指南》，同時歡迎關注筆者的微信公眾號：朱小廝的博客。

歡迎跳轉到本文的原文鏈接：https://honeypps.com/mq/kafka-lag-anymore/

---

在《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);

如此就可以達到與前面幾篇文章中關于獲取消費者詳情功能同樣的效果。這里有兩個注意要點：

PartitionAssignmentState中的coordinator是Node類型，這個類型需要自定義，Kafka原生的會報錯。

反序列化時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集合互轉的介紹會在下一篇文章中呈現。

歡迎跳轉到本文的原文鏈接：https://honeypps.com/mq/kafka-lag-anymore/

---

歡迎支持筆者新作：《深入理解Kafka:核心設計與實踐原理》和《RabbitMQ實戰指南》，同時歡迎關注筆者的微信公眾號：朱小廝的博客。

---

總結

以上是生活随笔為你收集整理的再看Kafka Lag的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。