溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

再看 Kafka Lag

發(fā)布時(shí)間:2020-07-14 07:03:17 來(lái)源:網(wǎng)絡(luò) 閱讀:596 作者:Java_老男孩 欄目:編程語(yǔ)言

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

可以直接將describeGroup返回的結(jié)果轉(zhuǎn)換成JSON然后傳至監(jiān)控頁(yè)面(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. 反序列化時(shí)忽略對(duì)象不存在的屬性
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
//3. 將Scala對(duì)象序列化成JSON字符串
String source = mapper.writeValueAsString(kafkaConsumerGroupService.describeGroup()._2.get());

這里需要采用的是jackson-module-scala的包實(shí)現(xiàn),如果直接用普通的JSON序列化方式那么會(huì)達(dá)不到想要的效果,jackson以及jackson-module-scala對(duì)應(yīng)的Maven庫(kù)如下:

<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版本不一致的話會(huì)報(bào)出一些異常。發(fā)散一下思維:既然可以序列化為JSON,那么完全可以通過(guò)JSON再反序列化會(huì)對(duì)象,只不過(guò)通過(guò)JSON作為中間媒介,將訪問受限的Scala對(duì)象轉(zhuǎn)變?yōu)镴ava對(duì)象,上面剩余代碼如下:

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

如此就可以達(dá)到與前面幾篇文章中關(guān)于獲取消費(fèi)者詳情功能同樣的效果。這里有兩個(gè)注意要點(diǎn):

  1. PartitionAssignmentState中的coordinator是Node類型,這個(gè)類型需要自定義,Kafka原生的會(huì)報(bào)錯(cuò)。
  2. 反序列化時(shí)Node會(huì)有一個(gè)empty的屬性不識(shí)別,解決方案參考代碼中的步驟2.

代碼更多細(xì)節(jié)請(qǐng)參考:代碼

通過(guò)JSON的序列化和反序列化操作實(shí)現(xiàn)了原本不能為之的事情,那么思維再發(fā)散一下,也可以序列化成字節(jié)流,比如通過(guò)ByteBuffer進(jìn)行轉(zhuǎn)換,只不過(guò)編程邏輯變得復(fù)雜了。

上面這段陳述有可能會(huì)讓人覺得Scala與Java之間的互操作起來(lái)不容易,其實(shí)不然,上面這段陳述只是用來(lái)補(bǔ)充一下如何獲取消費(fèi)者詳情的另一種方法,Scala與Java之間的互操作還是比較簡(jiǎn)單的,一般情況下都可以直接使用對(duì)方的類。對(duì)于集合而言,Scala中還有用于Scala與Java集合的互轉(zhuǎn)的scala.collection.JavaConverters(scala2.8.1開始引入),與此雷同的scala.collection.JavaConversions已被標(biāo)注為@Deprecated(since 2.12.0)。在scala代碼中如果需要集合轉(zhuǎn)換,首先引入scala.collection.JavaConverters._,進(jìn)而顯示調(diào)用asJava或者asScala方法完成轉(zhuǎn)型。關(guān)于Scala與Java集合互轉(zhuǎn)的介紹會(huì)在下一篇文章中呈現(xiàn)。


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

需要更詳細(xì)思維導(dǎo)圖和以下資料的可以加一下技術(shù)交流分享群:“708 701 457”免費(fèi)獲取

再看 Kafka Lag
再看 Kafka Lag
再看 Kafka Lag
再看 Kafka Lag

向AI問一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI