两种实现方式 | 如何查看消费者组的消费情况
版本：kafka_2.11-1.1.0
本文提供两种方式来查看消费者组的消费情况，分别通过命令行和 java api 的方式来消费 __consumer_offsets 。
一、通过命令行来查看消费情况
查看 kafka 消费者组列表：
./bin/kafka-consumer-groups.sh --bootstrap-server <kafka-ip>:9092 --list查看 kafka 中某一个消费者组的消费情况：
./bin/kafka-consumer-groups.sh --bootstrap-server <kafka-ip>:9092 --group test-17 --describetest-17 是消费者组。
在上面这张图中，我们可以看到该消费者组消费的 topic、partition、当前消费到的 offset 、最新 offset 、LAG(消费进度) 等等。如果消费者的 offset 很长时间没有提交导致 LAG 越来越大，则证明消费 Kafka 的服务异常。
消费者组消费 topic 的元数据信息，在旧版本里面是存储在 zookeeper 中，但由于 zookeeper 并不适合大批量的频繁写入操作，新版 kafka 已将消费者组的元数据信息保存在 kafka 内部的 topic 中，即 __consumer_offsets ，并提供了 kafka-console-consumer.sh 脚本供用户查看消费者组的元数据信息。
那么如何使用 kafka 提供的脚本查询某消费者组的元数据信息呢？
__consumer_offsets 默认有 50 个 partition，kafka 会根据 group.id 的 hash 值选择往哪个 partition 里面存放该 group 的元数据信息。计算 group.id 对应的 partition 的公式为：
Math.abs(groupID.hashCode()) % numPartitions举例：Math.abs("test-17".hashCode()) % 50，其中 test-17 是 group.id 。
找到 group.id 对应的 partition 后，就可以指定分区消费了：
./bin/kafka-console-consumer.sh --bootstrap-server message-1:9092 --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --partition 45kafka 0.11.0.0 版本(含)之前需要使用 formatter 为 kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter ， 0.11.0.0 版本以后(含)使用上面脚本中使用的 Class 。
脚本执行后输出的元数据信息有：
[test-17,topic-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());
}
}
}
}
只要消费者组提交位移，__consumer_offsets 里面就会增加对应的元数据信息。我们可以通过指定分区消费 __consumer_offsets ，来监控某消费者组的消费情况，避免在生产环境中消费程序假死而不自知。
小结
1、解析 __consumer_offsets 元数据信息需要设置 ByteArrayDeserializer 序列化和反序列化。
2、通过 kafka.coordinator.GroupMetadataManager 来解析元数据信息
3、__consumer_offsets 的元数据信息有两种：
- [key：OffsetKey，value：OffsetAndMetadata]：保存了消费者组各 partition 的 offset 元数据信息。
- [key：GroupMetadataKey，value：GroupMetadata]：保存了消费者组中各个消费者的信息。