0
0

【大数据学习 | kafka】kafka的偏移量管理

1. 偏移量的概念

消费者在消费数据的时候需要将消费的记录存储到一个位置，防止因为消费者程序宕机而引起断点消费数据丢失问题，下一次可以按照相应的位置从kafka中找寻数据，这个消费位置记录称之为偏移量offset。

kafka0.9以前版本将偏移量信息记录到zookeeper中

新版本中偏移量信息记录在__consumer_offsets中，这个topic是系统生成的，不仅仅帮助管理偏移量信息还能分配consumer给哪个coordinator管理，是一个非常重要的topic

它的记录方式和我们知道的记录方式一样 groupid + topic + partition ==> offset

其中存储到__consumer_offsets中的数据格式也是按照k-v进行存储的，其中k是groupid + topic + partition
value值为offset的偏移量信息。

  1. [hexuan@hadoop106 ~]$ kafka-topics.sh --bootstrap-server hadoop106:9092 --list
  2. __consumer_offsets
  3. topic_a
  4. topic_b
  5. topic_c
  6. topic_e
  7. topic_f
  8. topic_g

可以看到系统生成的topic

因为之前我们消费过很多数据，现在可以查看一下记录在这个topic中的偏移量信息

其中存在一个kafka-consumer-groups.sh 命令

  1. # 查看消费者组信息
  2. kafka-consumer-groups.sh --bootstrap-server hadoop106:9092 --list
  3. # 查询具体信息
  4. kafka-consumer-groups.sh --bootstrap-server hadoop106:9092 --describe --group my-group
  5. # 查看活跃信息
  6. kafka-consumer-groups.sh --bootstrap-server hadoop106:9092 --describe --group my-group --members

查看消费者组信息：

  1. [hexuan@hadoop106 ~]$ kafka-consumer-groups.sh --bootstrap-server hadoop106:9092 --list
  2. hainiu_group
  3. hainiu_group2

当前使用组信息：

  1. [hexuan@hadoop106 ~]$ kafka-consumer-groups.sh --bootstrap-server hadoop106:9092 --describe --group hainiu_group
  3. GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                  HOST            CLIENT-ID
  4. hainiu_group    topic_c         0          0               0               0               consumer-hainiu_group-1-41a9ebd6-99a3-4d83-b1d7-88a2a9295054 /192.168.154.1  consumer-hainiu_group-1
  5. hainiu_group    topic_b         1          1438            1438            0               -                                                            -               -
  6. hainiu_group    topic_b         0          1440            1440            0               -                                                            -               -
  7. hainiu_group    topic_b         3          1417            1417            0               -                                                            -               -
  8. hainiu_group    topic_b         4          1473            1473            0               -                                                            -               -
  9. hainiu_group    topic_b         5          1440            1440            0               -                                                            -               -
  10. hainiu_group    topic_b         2          1407            1407            0               -                                                            -               -
  11. hainiu_group    topic_b         6          1391            1391            0               -

当前组消费偏移量信息：

  1. GROUP：组名 
  2. TOPIC：topic信息           
  3. PARTITION：分区  
  4. CURRENT-OFFSET：当前消费偏移量  
  5. LOG-END-OFFSET：这个分区总共存在多少数据  
  6. LAG：还差多少没消费             
  7. CONSUMER-ID：随机消费者id                                                  
  8. HOST：主机名            
  9. CLIENT-ID：客户端id

同时我们也可以查询**__consumer_offset中的原生数据：**

  1. kafka-console-consumer.sh  --bootstrap-server hadoop106:9092 \
  2. --topic __consumer_offsets --from-beginning --formatter \
  3. kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter

使用元数据格式化方式查看偏移量信息数据

key展示的是groupid,topic,partition , value值展示的是当前的偏移量信息

并且在这个topic中是追加形式一致往里面写入的

2. 偏移量的自动管理

那么我们已经看到了偏移量的存储但是偏移量究竟是怎么提交的呢？

首先我们没有设置任何的偏移量提交的代码，这个是默认开启的，其中存在两个参数

  1. pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
  2. //开启自动提交偏移量信息
  3. pro.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  4. //默认提交间隔5s

官网的设置参数为两个true和5000。

所以我们在没有开启默认提交的时候已经自动提交了

为了演示自动提交的效果我们引入一个参数

  1. auto.offset.reset

这个参数用于控制没有偏移量存储的时候，应该从什么位置进行消费数据

（因为偏移量自动提交默认是5秒一次，如果数据在5秒内消费完毕，则会造成偏移量并没有存储的情况）

其中参数值官网中给出三个

  1. [latest, earliest, none]
  3. latest:从最新位置消费
  5. earliest：最早位置消费数据
  7. none：如果不指定消费的偏移量直接报错

一定要记得一点，如果有偏移量信息那么以上的设置是无效的.

官方文档显示给出的该参数的默认值为lastest，即从最新位置开始消费。

现在我们设置读取位置为最早位置，并且消费数据，看看可不可以记录偏移量，断点续传

思路：

首先修改组id为一个新的组，然后从最早位置消费数据，如果记录了偏移量，那么重新启动消费者会看到，没有任何数据，因为之前记录了消费数据的位置

整体代码如下：

  1. package com.hainiu.kafka;
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5. import org.apache.kafka.common.serialization.StringDeserializer;
  7. import java.time.Duration;
  8. import java.util.*;
  10. public class Consumer1 {
  11.     public static void main(String[] args) {
  12.         Properties pro = new Properties();
  13.         pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"nn1:9092");
  14.         pro.put(ConsumerConfig.GROUP_ID_CONFIG,"new_group");
  15.         pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  16.         pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  17.         pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
  18.         pro.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  19.         pro.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  21.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
  22.         List<String> topics = Arrays.asList("topic_d","topic_e");
  23.         consumer.subscribe(topics);
  25.         while (true){
  26.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  27.             Iterator<ConsumerRecord<String, String>> it = records.iterator();
  28.             while(it.hasNext()){
  29.                 ConsumerRecord<String, String> record = it.next();
  30.                 System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
  31.             }
  32.         }
  33.     }
  34. }

运行完毕打印数据

这个时候我们需要在5s之内关闭应用，然后重新启动，因为提交的间隔时间是5s

再次启动

我们发现数据依旧被消费出来了，证明之前的偏移量存储没有任何效果和作用，因为间隔时间是5s

现在我们等待5s后在关闭应用

发现没有任何数据产生，因为偏移量已经提交了

3. 偏移量的手动提交

如上的案例我们发现偏移量的管理如果交给系统自己管理，我们没有办法及时的修改和管理偏移量信息，这个时候我们需要手动来提交给管理偏移量，更加及时和方便

这个时候引入两个方法

  1. consumer.commitAsync();
  2. consumer.commitSync();

commitAsync 异步提交方式：只提交一次，不管成功与否不会重试

commitSync 同步提交方式：同步提交方式会一直提交到成功为止

一般我们都会选择异步提交方式，他们的功能都是将拉取到的一整批数据的最大偏移量直接提交到__consumer_offsets中，但是同步方式会很浪费资源，异步方式虽然不能保证稳定性但是我们的偏移量是一直递增存储的，所以偶尔提交不成功一个两个不影响我们的使用

  1.  pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  2. //设定自动提交为false
  3. consumer.commitSync();
  4. consumer.commitAsync();
  5. //设定提交方式为手动提交

整体代码如下：

  1. package com.hainiu.kafka.consumer;
  3. /**
  4.  * ClassName : consumer_offsets
  5.  * Package : com.hainiu.kafka.consumer
  6.  * Description
  7.  *
  8.  * @Author HeXua
  9.  * @Create 2024/11/5 21:30
  10.  * Version 1.0
  11.  */
  13. import org.apache.kafka.clients.consumer.*;
  14. import org.apache.kafka.common.TopicPartition;
  15. import org.apache.kafka.common.serialization.StringDeserializer;
  17. import java.time.Duration;
  18. import java.util.*;
  20. public class Consumer_CommitSync {
  21.     public static void main(String[] args) {
  22.         Properties pro = new Properties();
  23.         pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop106:9092");
  24.         pro.put(ConsumerConfig.GROUP_ID_CONFIG,"hainiu_group2");
  25.         pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  26.         pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  27.         pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  28.         pro.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  29. //        pro.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  31.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
  32.         List<String> topics = Arrays.asList("topic_h");
  33.         consumer.subscribe(topics);
  35.         while (true){
  36.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  37.             Iterator<ConsumerRecord<String, String>> it = records.iterator();
  38.             while(it.hasNext()){
  39.                 ConsumerRecord<String, String> record = it.next();
  40.                 System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
  41.             }
  42.             consumer.commitAsync();
  43. //            consumer.commitSync();
  44.         }
  45.     }
  46. }

现在先在topic中输入部分数据

然后启动消费者，当存在数据打印的时候马上关闭掉应用，在此启动会发现数据不会重新消费

  1. topic_h->5->12->null->1
  2. topic_h->5->13->null->2
  3. topic_h->5->14->null->3
  4. topic_h->5->15->null->4
  5. topic_h->5->16->null->5
  6. topic_h->5->17->null->6

偏移量已经提交不会重复消费数据

4. 断点消费数据

在没有偏移量的时候我们可以设定

auto.offset.reset进行数据的消费

可选参数有 latest earliest none等位置

但是如果存在偏移量以上的设定就不在好用了，我们需要根据偏移量的位置进行断点消费数据

但是有的时候我们需要指定位置消费相应的数据

这个时候我们需要使用到

  1. consumer.seek();
  2. //可以指定位置进行数据的检索

但是我们不能随意的指定消费者消费数据的位置，因为在启动消费者的时候，一个组中会存在多个消费者，每个人拿到的对应分区是不同的，所以我们需要知道这个消费者能够获取的分区是哪个，然后再指定相应的断点位置

这里我们就需要监控分区的方法展示出来所有订阅的分区信息

  1.  consumer.subscribe(topics, new ConsumerRebalanceListener() {
  2.             @Override
  3.             public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
  5.             }
  7.             @Override
  8.             public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
  10.             }
  11.         });

为了演示效果我们使用生产者在topic_d中增加多个消息

  1. package com.hainiu.kafka.consumer;
  3. /**
  4.  * ClassName : Producer2
  5.  * Package : com.hainiu.kafka.consumer
  6.  * Description
  7.  *
  8.  * @Author HeXua
  9.  * @Create 2024/11/5 23:01
  10.  * Version 1.0
  11.  */
  12. import org.apache.kafka.clients.producer.KafkaProducer;
  13. import org.apache.kafka.clients.producer.ProducerConfig;
  14. import org.apache.kafka.clients.producer.ProducerRecord;
  15. import org.apache.kafka.common.serialization.StringSerializer;
  17. import java.util.Properties;
  19. public class Producer2 {
  20.     public static void main(String[] args) {
  21.         Properties pro = new Properties();
  22.         pro.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop106:9092");
  23.         pro.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  24.         pro.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  25.         KafkaProducer<String, String> producer = new KafkaProducer<String, String>(pro);
  26.         for (int i = 0; i < 1000; i++) {
  27.             ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic_d", "" + i, "message"+i);
  28.             producer.send(record);
  29.         }
  30.         producer.close();
  31.     }
  32. }

随机发送数据到不同的节点，使用随机key

然后使用断点消费数据

不设置任何的偏移量提交操作和断点位置

  1. package com.hainiu.kafka.consumer;
  3. /**
  4.  * ClassName : ConsumerWithUDOffset
  5.  * Package : com.hainiu.kafka.consumer
  6.  * Description
  7.  *
  8.  * @Author HeXua
  9.  * @Create 2024/11/5 23:03
  10.  * Version 1.0
  11.  */
  12. import org.apache.kafka.clients.consumer.*;
  13. import org.apache.kafka.common.TopicPartition;
  14. import org.apache.kafka.common.serialization.StringDeserializer;
  16. import java.time.Duration;
  17. import java.util.*;
  19. public class ConsumerWithUDOffset {
  20.     public static void main(String[] args) {
  21.         Properties pro = new Properties();
  22.         pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop106:9092");
  23.         pro.put(ConsumerConfig.GROUP_ID_CONFIG,"new1");
  24.         pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  25.         pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  26.         pro.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
  27.         pro.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,6000);
  28.         pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
  30.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
  31.         List<String> topics = Arrays.asList("topic_h");
  32.         // range roundRobin sticky cooperativeSticky
  33.         consumer.subscribe(topics, new ConsumerRebalanceListener() {
  34.             @Override
  35.             public void onPartitionsRevoked(Collection<TopicPartition> collection) {
  37.             }
  39.             @Override
  40.             public void onPartitionsAssigned(Collection<TopicPartition> collection) {
  41.                 for (TopicPartition topicPartition : collection) {
  42.                     consumer.seek(topicPartition,195);
  43.                 }
  44.             }
  45.         });
  47.         while (true){
  48.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  49.             Iterator<ConsumerRecord<String, String>> it = records.iterator();
  50.             while(it.hasNext()){
  51.                 ConsumerRecord<String, String> record = it.next();
  52.                 System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
  53.             }
  54.             consumer.commitAsync();
  55.         }
  56.     }
  57. }

5. 时间断点

kafka没有给大家提供直接根据时间找到断点位置的方法，我们需要根据时间找到偏移量，然后根据偏移量进行数据消费

  1. consumer.offsetsForTimes();
  2. //通过这个方法找到对应时间的偏移量位置
  3. consumer.seek();
  4. //然后在通过这个方法根据断点进行消费数据

整体代码如下

  1. package com.hainiu.kafka;
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5. import org.apache.kafka.common.serialization.StringDeserializer;
  7. import java.time.Duration;
  8. import java.util.*;
  10. public class Consumer1 {
  11.     public static void main(String[] args) {
  12.         Properties pro = new Properties();
  13.         pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"nn1:9092");
  14.         pro.put(ConsumerConfig.GROUP_ID_CONFIG,"new_group221");
  15.         pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  16.         pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  17.         pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
  19.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
  20.         List<String> topics = Arrays.asList("topic_e");
  21.         consumer.subscribe(topics, new ConsumerRebalanceListener() {
  22.             @Override
  23.             public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
  24.                 // no op
  25.             }
  27.             @Override
  28.             public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
  29.                 HashMap<TopicPartition, Long> map = new HashMap<>();
  30.                 for (TopicPartition partition : partitions) {
  31.                     map.put(partition,1675076400000L);
  32.                     //将时间和分区绑定在一起，然后合并在一起放入到检索方法中
  33.                 }
  34.                 Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(map);
  35.                 //根据时间获取时间对应的偏移量位置
  36.                 for (Map.Entry<TopicPartition, OffsetAndTimestamp> en : offsets.entrySet()) {
  37.                     System.out.println(en.getKey()+"-->"+en.getValue());
  38.                     if(en.getValue() != null){
  39.                         consumer.seek(en.getKey(),en.getValue().offset());
  40.                         //获取每个分区的偏移量的位置，使用seek进行找寻数据
  41.                     }
  42.                 }
  44.             }
  45.         });
  47.         while (true){
  48.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
  49.             Iterator<ConsumerRecord<String, String>> it = records.iterator();
  50.             while(it.hasNext()){
  51.                 ConsumerRecord<String, String> record = it.next();
  52.                 System.out.println(record.topic()+"->"+record.partition()+"->"+ record.offset()+"->"+record.key()+"->"+record.value());
  53.             }
  54. //            consumer.commitAsync();
  55.         }
  56.     }
  57. }
标签： 大数据 sqlite oracle
本文转载自: https://blog.csdn.net/2301_80912559/article/details/143525859
版权归原作者 Vez'nan的幸福生活 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【大数据学习 | kafka】kafka的偏移量管理”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。