如何查看消费者组的消费情况

kafka_2.11-1.1.0

本文提供两种方式来查看消费者组的消费情况，分别通过命令行和 java api 的方式来消费 __consumer_offsets 。

一、通过命令行来查看消费情况

查看 kafka 消费者组列表：

1	./bin/kafka-consumer-groups.sh --bootstrap-server <kafka-ip>:9092 --list

查看 kafka 中某一个消费者组的消费情况：

1	./bin/kafka-consumer-groups.sh --bootstrap-server <kafka-ip>:9092 --group test-17 --describe

test-17 是消费者组。

在上面这张图中，我们可以看到该消费者组消费的 topic、partition、当前消费到的 offset 、最新 offset 、LAG(消费进度) 等等。如果消费者的 offset 很长时间没有提交导致 LAG 越来越大，则证明消费 Kafka 的服务异常。

消费者组消费 topic 的元数据信息，在旧版本里面是存储在 zookeeper 中，但由于 zookeeper 并不适合大批量的频繁写入操作，新版 kafka 已将消费者组的元数据信息保存在 kafka 内部的 topic 中，即 __consumer_offsets topic ，并提供了 kafka-console-consumer.sh 脚本供用户查看消费者组的元数据信息。

那么如何使用 kafka 提供的脚本查询某消费者组的元数据信息呢？

__consumer_offsets 默认有 50 个 partition，kafka 会根据 group.id 的 hash 值选择往哪个 partition 里面存放该 group 的元数据信息。计算 group.id 对应的 partition 的公式为：

1	Math.abs(groupID.hashCode()) % numPartitions

举例：Math.abs(“test-17”.hashCode()) % 50，其中 test-17 是 group.id 。

找到 group.id 对应的 partition 后，就可以指定分区消费了：

1	./bin/kafka-console-consumer.sh --bootstrap-server message-1:9092 --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --partition 45

kafka 0.11.0.0 版本(含)之前需要使用 formatter 为 kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter ， 0.11.0.0 版本以后(含)使用上面脚本中使用的 Class 。

脚本执行后输出的元数据信息有：

1	[test-17,history-event-test,0]::[OffsetMetadata[36672,NO_METADATA],CommitTime 1599746660064,ExpirationTime 1599833060064]

[消费者组 : 消费的topic : 消费的分区] :: [offset位移], [offset提交时间], [元数据过期时间]

二、通过 java api 来查看消费情况，方便做告警监控

使用 java api 来查看 __consumer_offsets 元数据信息，更加灵活方便。比如我就用了以下方式做了消费者组消费的告警监控。

pom 依赖：

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>1.1.0</version>
    <exclusions>
        <exclusion>
            <artifactId>kafka-clients</artifactId>
            <groupId>org.apache.kafka</groupId>
        </exclusion>
        <exclusion>
            <artifactId>slf4j-log4j12</artifactId>
            <groupId>org.slf4j</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.1.0</version>
</dependency>

消费者配置：

__consumer_offsets 中的数据需要用字节来转码，所以消费者配置中需要设置 ByteArrayDeserializer 序列化和反序列化：

props.put("key.deserializer",
        "org.apache.kafka.common.serialization.ByteArrayDeserializer");
props.put("value.deserializer",
        "org.apache.kafka.common.serialization.ByteArrayDeserializer");

消费者示例代码：

/**
     * 指定分区消费__consumer_offsets
     */
public void consumerTopic() {
    consumer.assign(Arrays.asList(new TopicPartition("__consumer_offsets", 45)));
    ConsumerRecords<byte[], byte[]> records;

    while (true) {
        records = consumer.poll(500);

        for (ConsumerRecord<byte[], byte[]> record : records) {
            BaseKey key = GroupMetadataManager.readMessageKey(ByteBuffer.wrap(record.key()));
            if (key instanceof OffsetKey) {
                GroupTopicPartition partition = (GroupTopicPartition) key.key();
                String topic = partition.topicPartition().topic();
                String group = partition.group();

                log.info("consumer group: [{}], consumer topic: [{}], key: [{}]", group, topic, key.toString());
            } else if (key instanceof GroupMetadataKey){
                log.info("groupMetadataKey: [{}]", key.key());    
				//第一个参数为group id，先将key转换为GroupMetadataKey类型，再调用它的key()方法就可以获得group id
    			GroupMetadata groupMetadata = GroupMetadataManager.readGroupMessageValue(((GroupMetadataKey) key).key(), ByteBuffer.wrap(record.value()));
				log.info("GroupMetadata: [{}]", groupMetadata.toString());
            }
        }
    }
}