如何獲取Kafka的消費者詳情——從Scala到Java的切換

前文摘要

在前面的文章《Kafka的Lag計算誤區(qū)及正確實現(xiàn)》中介紹了如何計算消費者的消費滯后量(Lag)，并且講解了如何調用Kafka的kafka.admin.ConsumerGroupCommand文件中的KafkaConsumerGroupService來發(fā)送OffsetRequest和OffsetFetchRequest兩個請求，進而通過兩個請求結果之間的差值來獲得結果。不過如果你不想修改kafka-core的代碼并重新編譯的話，這種實現(xiàn)方式無法成功，所以本文的主要目的就是通過調用更底層的API來實現(xiàn)不修改kafka-core的代碼來實現(xiàn)KafkaConsumerGroupService的功能，即通過Java調用Scala的代碼來實現(xiàn)獲取Kafka的消費者詳情的功能。

成都創(chuàng)新互聯(lián)是創(chuàng)新、創(chuàng)意、研發(fā)型一體的綜合型網站建設公司，自成立以來公司不斷探索創(chuàng)新，始終堅持為客戶提供滿意周到的服務，在本地打下了良好的口碑，在過去的十年時間我們累計服務了上千家以及全國政企客戶，如辦公窗簾等企業(yè)單位，完善的項目管理流程，嚴格把控項目進度與質量監(jiān)控加上過硬的技術實力獲得客戶的一致夸獎。

目標及實現(xiàn)

實現(xiàn)如同 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_ID

KafkaConsumerGroupService的核心方法是CollectGroupAssignment，其方法參數(shù)為一個consumer group的groupId，方法輸出為上面示例中的列表信息。CollectGroupAssignment方法主要有以下幾個步驟：

根據(jù)groupId調用describeConsumerGroup方法(內部原理是發(fā)送DescribeGroupsRequest請求)來獲取consumer group的基本信息，參考上面示例中的CONSUMER-ID、HOST、CLIENT-ID以及TopicPartition信息，但是沒有CURRENT-OFFSET、LOG-END-OFFSET、LAG信息。注意這里的LOG-END-OFFSET是消費者可見的LEO，不是生產者可見的LEO，也就是通俗意義上的HW。
根據(jù)groupId調用listGroupOffsets方法(內部原理是發(fā)送OffsetFetchRequest請求)來獲取各個分區(qū)(Partition)的對應的消費位移CURRENT-OFFSET。
通過調用KafkaConsumer的endOffsets方法來獲取TopicPartition對應的HW，即示例中的LOG-END-OFFSET。
計算Lag并組合成信息列表List<partitionassignmentstate style="margin-block-start: 0px; margin-block-end: 0px; margin: 0px; padding: 0px;">。</partitionassignmentstate>

改造

對應Java版的KafkaConsumerGroupService改造代碼可以參見代碼，目錄結構如下圖所示：
如何獲取 Kafka 的消費者詳情 —— 從 Scala 到 Java 的切換

其中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節(jié)點信息
    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中包含了眾多管理類方法，主要是通過發(fā)送各種自定義協(xié)議請求來完成，上面步驟中所說的describeConsumerGroup和listGroupOffsets方法也是通過AdminClient來實現(xiàn)的；KafkaConsumer主要是用來獲取TopicPartition對應的HW(消費者可見的LogEndOffsets)的。

KafkaConsumerGroupCustomService中與scala版對應的collectGroupAssignment方法如下（詳細步驟參考代碼注釋）：

public List<PartitionAssignmentState> collectGroupAssignment(
        AdminClient adminClient, KafkaConsumer<String, String> consumer,
        String group) {
    //1. 獲取consumer group的基本信息，包括CONSUMER-ID、HOST、
    // CLIENT-ID以及TopicPartition信息
    AdminClient.ConsumerGroupSummary consumerGroupSummary
            = adminClient.describeConsumerGroup(group, 0);
    List<TopicPartition> assignedTopicPartitions = new ArrayList<>();
    List<PartitionAssignmentState> rowsWithConsumer = new ArrayList<>();
    scala.collection.immutable.List<AdminClient.ConsumerSummary> consumers
            = consumerGroupSummary.consumers().get();
    if (consumers != null) {
        //2. 獲取各個分區(qū)(Partition)的對應的消費位移CURRENT-OFFSET
        scala.collection.immutable.Map<TopicPartition, Object> offsets
                = adminClient.listGroupOffsets(group);
        if (offsets.nonEmpty()) {
            String state = consumerGroupSummary.state();
            // 3. 還有一個狀態(tài)是Dead表示"group"對應的consumer group不存在
            if (state.equals("Stable") || state.equals("Empty")
                    || state.equals("PreparingRebalance")
                    || state.equals("AwaitingSync")) {
                List<ConsumerSummary> consumerList = changeToJavaList(consumers);
                // 4. 獲取當前有消費者的消費信息，即包含CONSUMER-ID、HOST、CLIENT-ID
                rowsWithConsumer = getRowsWithConsumer(consumerGroupSummary, offsets,
                        consumer, consumerList, assignedTopicPartitions, group);
            }
        }
        //5. 獲取當前沒有消費者的消費信息
        List<PartitionAssignmentState> rowsWithoutConsumer =
                getRowsWithoutConsumer(consumerGroupSummary,
                offsets, consumer, assignedTopicPartitions, group);
        //6. 合并結果
        rowsWithConsumer.addAll(rowsWithoutConsumer);
    }
    return rowsWithConsumer;
}

KafkaConsumerGroupCustomService類中包含有getRowsWithConsumer()、getRowsWithoutConsumer()、changeToJavaList等私有方法也都是在Scala語言與Java語言之間進行切換，這樣可以不需要修改kafka-core的原生代碼而通過外部的封裝調用既可以實現(xiàn)獲取Kafka消費者詳情的功能。光看代碼比較抽象，建議對此感興趣的同學可以親自對比一下kafka-core包中kafka.admin.ConsumerGroupCommand的KafkaConsumerGroupSerivice與筆者自定義的KafkaConsumerGroupCustomService的實現(xiàn)來了解下Scala語言到Java語言的轉換。

如果需要打印詳情可以調用KafkaConsumerGroupCustomService同目錄的ConsumerGroupUtils類中的printPasList(List list)方法。注意要運行這些代碼需要JDK8的環(huán)境，筆者為了讓代碼顯得“騷氣”一點就用來一點Java8的語法，如果需要Java7的代碼實現(xiàn)可以關注私聊。

或許有些同學對于Scala和Java交叉的代碼并不感冒，想要尋求一種存Java式的實現(xiàn)方式，那么在這里怎么實現(xiàn)呢？答案是通過KafkaAdminClient，它是AdminClient的Java版實現(xiàn)，從Kafka0.11.0.0版本開始引入的，不過KafkaAdminClient本身并沒有提供describeConsumerGroup、listGroupOffsets之類的方法給我們直接使用，擴展一下也很方便，由于篇幅限制，這部分的內容將在下一篇文章中進行介紹，如果想要先一睹為快，可以參考下代碼實現(xiàn)，詳細的邏輯解析敬請期待….

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

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

如何獲取 Kafka 的消費者詳情 —— 從 Scala 到 Java 的切換

當前文章：如何獲取Kafka的消費者詳情——從Scala到Java的切換
文章位置：http://muchs.cn/article8/jpisop.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供、軟件開發(fā)、網站建設、網站排名、全網營銷推廣、響應式網站

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

猜你還喜歡下面的內容