在《Kafka的Lag計算誤區(qū)及正確實現(xiàn)》一文中提及了kafka.admin.ConsumerGroupCommand.PartitionAssignmentState無法被外部訪問，故要將PartitionAssignmentState前的protected修飾符去掉

溪湖ssl適用于網(wǎng)站、小程序/APP、API接口等需要進行數(shù)據(jù)傳輸應用場景，ssl證書未來市場廣闊！成為創(chuàng)新互聯(lián)公司的ssl證書銷售渠道，可以享受市場價格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：13518219792（備注：SSL證書合作）期待與您的合作！

可以直接將describeGroup返回的結果轉換成JSON然后傳至監(jiān)控頁面（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的包實現(xiàn)，如果直接用普通的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版本不一致的話會報出一些異常。發(fā)散一下思維：既然可以序列化為JSON，那么完全可以通過JSON再反序列化會對象，只不過通過JSON作為中間媒介，將訪問受限的Scala對象轉變?yōu)镴ava對象，上面剩余代碼如下：

//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.

代碼更多細節(jié)請參考：代碼

通過JSON的序列化和反序列化操作實現(xiàn)了原本不能為之的事情，那么思維再發(fā)散一下，也可以序列化成字節(jié)流，比如通過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._，進而顯示調(diào)用asJava或者asScala方法完成轉型。關于Scala與Java集合互轉的介紹會在下一篇文章中呈現(xiàn)。

---

本文的重點是你有沒有收獲與成長，其余的都不重要，希望讀者們能謹記這一點。同時我經(jīng)過多年的收藏目前也算收集到了一套完整的學習資料，包括但不限于：分布式架構、高可擴展、高性能、高并發(fā)、Jvm性能調(diào)優(yōu)、Spring，MyBatis，Nginx源碼分析，Redis，ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多個知識點高級進階干貨，希望對想成為架構師的朋友有一定的參考和幫助

需要更詳細思維導圖和以下資料的可以加一下技術交流分享群：“708 701 457”免費獲取

創(chuàng)新互聯(lián)www.cdcxhl.cn，專業(yè)提供香港、美國云服務器，動態(tài)BGP最優(yōu)骨干路由自動選擇，持續(xù)穩(wěn)定高效的網(wǎng)絡助力業(yè)務部署。公司持有工信部辦法的idc、isp許可證， 機房獨有T級流量清洗系統(tǒng)配攻擊溯源，準確進行流量調(diào)度，確保服務器高可用性。佳節(jié)活動現(xiàn)已開啟，新人活動云服務器買多久送多久。

文章名稱：再看KafkaLag-創(chuàng)新互聯(lián)
文章出自：http://aaarwkj.com/article38/cchipp.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供服務器托管、App開發(fā)、網(wǎng)站設計公司、手機網(wǎng)站建設、商城網(wǎng)站、全網(wǎng)營銷推廣

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)