两种实现方式 |
您所在的位置：网站首页 › 怎么看苹果id的消费记录清单 › 两种实现方式 |
两种实现方式
版本：kafka_2.11-1.1.0 本文提供两种方式来查看消费者组的消费情况，分别通过命令行和 java api 的方式来消费 __consumer_offsets 。 一、通过命令行来查看消费情况查看 kafka 消费者组列表： 代码语言：javascript复制./bin/kafka-consumer-groups.sh --bootstrap-server :9092 --list查看 kafka 中某一个消费者组的消费情况： 代码语言：javascript复制./bin/kafka-consumer-groups.sh --bootstrap-server :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 的公式为： 代码语言：javascript复制Math.abs(groupID.hashCode()) % numPartitions举例：Math.abs("test-17".hashCode()) % 50，其中 test-17 是 group.id 。 找到 group.id 对应的 partition 后，就可以指定分区消费了： 代码语言：javascript复制./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 。 脚本执行后输出的元数据信息有： 代码语言：javascript复制[test-17,topic-test,0]::[OffsetMetadata[36672,NO_METADATA],CommitTime 1599746660064,ExpirationTime 1599833060064][消费者组 : 消费的topic : 消费的分区] :: [offset位移], [offset提交时间], [元数据过期时间] 二、通过 java api 来查看消费情况，方便做告警监控使用 java api 来查看 __consumer_offsets 元数据信息，更加灵活方便。比如我就用了以下方式做了消费者组消费的告警监控。pom 依赖： 代码语言：javascript复制 org.apache.kafka kafka_2.11 1.1.0 kafka-clients org.apache.kafka slf4j-log4j12 org.slf4j org.apache.kafka kafka-clients 1.1.0消费者配置： __consumer_offsets 中的数据需要用字节来转码，所以消费者配置中需要设置 ByteArrayDeserializer 序列化和反序列化： 代码语言：javascript复制props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");消费者示例代码： 代码语言：javascript复制/** * 指定分区消费__consumer_offsets */ public void consumerTopic() { consumer.assign(Arrays.asList(new TopicPartition("__consumer_offsets", 45))); ConsumerRecords records; while (true) { records = consumer.poll(500); for (ConsumerRecord 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]：保存了消费者组中各个消费者的信息。 |
【本文地址】
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |