devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka 消费者进行消费数据的各种场景的API(你值得一看)_kafkaconsumer 的所有api

作者：笔触狂放9 | 2024-05-28 19:21:18

赞

踩

kafkaconsumer 的所有api

一 kafka消费端的参数

 

二 实现案例

2.1 订阅某个主题

创建一个独立消费者，消费 kafka-ljf 主题中数据。
注意： 在消费者 API 代码中必须配置消费者组 id 。命令行启动消费者不填写消费者组
id 会被自动填写随机的消费者组 id 。
2.消费者代码 
  1. package com.ljf.spring.boot.demo.consumer;
  2.  
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.common.serialization.StringDeserializer;
  8.  
  9. import java.time.Duration;
  10. import java.util.ArrayList;
  11. import java.util.Properties;
  12.  
  13. /**
  14.  * @ClassName: ConsumerTopicDemo
  15.  * @Description: TODO
  16.  * @Author: liujianfu
  17.  * @Date: 2022/04/10 14:02:05
  18.  * @Version: V1.0
  19.  **/
  20. public class ConsumerTopicDemo {
  21.     public static void main(String[] args) {
  22.         // 1.创建消费者的配置对象
  23.         Properties properties = new Properties();
  24.         // 2.给消费者配置对象添加参数
  25.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  26.                 "192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");
  27.         // 配置序列化 必须
  28.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  29.                 StringDeserializer.class.getName());
  30.  
  31.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  32.                 StringDeserializer.class.getName());
  33.         // 配置消费者组（组名任意起名） 必须
  34.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "beijing");
  35.         // 创建消费者对象
  36.         KafkaConsumer<String, String> kafkaConsumer = new
  37.                 KafkaConsumer<String, String>(properties);
  38.         // 注册要消费的主题（可以消费多个主题）
  39.         ArrayList<String> topics = new ArrayList<>();
  40.         topics.add("kafka-ljf");
  41.         kafkaConsumer.subscribe(topics);
  42.         // 拉取数据打印
  43.         while (true) {
  44.             // 设置 1s 中消费一批数据
  45.             ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
  46.             // 打印消费到的数据
  47.             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
  48.                 System.out.println(consumerRecord);
  49.             }
  50.         }
  51.     }
  52.  
  53. }

3.执行生产者产生数据

 4.消费数据，观察

2.2 订阅某个主题下的某个分区

需求：创建一个独立消费者，消费 kafka-ljf主题 0 号分区的数据。

2.代码

 

  1. package com.ljf.spring.boot.demo.consumer;
  2.  
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.common.TopicPartition;
  8. import org.apache.kafka.common.serialization.StringDeserializer;
  9.  
  10. import java.time.Duration;
  11. import java.util.ArrayList;
  12. import java.util.Properties;
  13.  
  14. /**
  15.  * @ClassName: ConsumerPartitionDemo
  16.  * @Description: TODO
  17.  * @Author: liujianfu
  18.  * @Date: 2022/04/10 14:55:31
  19.  * @Version: V1.0
  20.  **/
  21. public class ConsumerPartitionDemo {
  22.     public static void main(String[] args) {
  23.         // 1.创建消费者的配置对象
  24.         Properties properties = new Properties();
  25.         // 2.给消费者配置对象添加参数
  26.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  27.                 "192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");
  28.         // 配置序列化 必须
  29.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  30.                 StringDeserializer.class.getName());
  31.  
  32.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  33.                 StringDeserializer.class.getName());
  34.         // 配置消费者组（必须），名字可以任意起
  35.         properties.put(ConsumerConfig.GROUP_ID_CONFIG,"beijing");
  36.         KafkaConsumer<String, String> kafkaConsumer = new
  37.                 KafkaConsumer<>(properties);
  38.         // 消费某个主题的某个分区数据，0号分区
  39.         ArrayList<TopicPartition> topicPartitions = new
  40.                 ArrayList<>();
  41.         topicPartitions.add(new TopicPartition("kafka-ljf", 0));
  42.         kafkaConsumer.assign(topicPartitions);
  43.         while (true){
  44.             ConsumerRecords<String, String> consumerRecords =
  45.                     kafkaConsumer.poll(Duration.ofSeconds(1));
  46.             for (ConsumerRecord<String, String> consumerRecord :
  47.                     consumerRecords) {
  48.                 System.out.println(consumerRecord);
  49.             }
  50.         }
  51.     }
  52. }

 3.生产者生产数据

 4.消费者消费

可见只消费了0号分区上的数据 

2.3  消费者组案例

测试同一个主题的分区数据，只能由一个消费者组中的一个消费。

1.consumer代码复制一份，变为两个消费者

 

2. 消费者2：

 3.生产者：

 

4.查看消费者消费信息

 5.查看消费者2消费信息

 结论：即可看到两个消费者在消费不同 分区的数据。消费者一消费分区1的数据，消费者2消费分区2的数据。

2.4  指定offset消费

auto.offset.reset = earliest | latest | none   其中 默认是 latest 。
当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量
时（例如该数据已被删除），该怎么办？
（ 1 ） earliest ：自动将偏移量重置为最早的偏移量，相当于   --from-beginning 。
（2） latest （默认值） ：自动将偏移量重置为最新偏移量。
（3） none ：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

 （4）任意指定 offset 位移开始消费

代码：

具体代码：

  1. package com.ljf.spring.boot.demo.consumer;
  2.  
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.common.TopicPartition;
  8. import org.apache.kafka.common.serialization.StringDeserializer;
  9.  
  10. import java.time.Duration;
  11. import java.util.ArrayList;
  12. import java.util.HashSet;
  13. import java.util.Properties;
  14. import java.util.Set;
  15.  
  16. /**
  17.  * @ClassName: ConsumerSeekDemo
  18.  * @Description: TODO
  19.  * @Author: liujianfu
  20.  * @Date: 2022/04/10 16:08:01
  21.  * @Version: V1.0
  22.  **/
  23. public class ConsumerSeekDemo {
  24.     public static void main(String[] args) {
  25.         // 0 配置信息
  26.         Properties properties = new Properties();
  27.         // 连接
  28.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  29.                 "192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");
  30.         // key value 反序列化
  31.  
  32.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  33.                 StringDeserializer.class.getName());
  34.  
  35.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  36.                 StringDeserializer.class.getName());
  37.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "beijing");
  38.         // 1 创建一个消费者
  39.         KafkaConsumer<String, String> kafkaConsumer = new
  40.                 KafkaConsumer<>(properties);
  41.         // 2 订阅一个主题
  42.         ArrayList<String> topics = new ArrayList<>();
  43.         topics.add("kafka-ljf");
  44.         kafkaConsumer.subscribe(topics);
  45.         Set<TopicPartition> assignment= new HashSet<>();
  46.         while (assignment.size() == 0) {
  47.             kafkaConsumer.poll(Duration.ofSeconds(1));
  48.             // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
  49.             assignment = kafkaConsumer.assignment();
  50.         }
  51.         // 遍历所有分区，并指定 offset 从 10 的位置开始消费
  52.         for (TopicPartition tp: assignment) {
  53.             kafkaConsumer.seek(tp, 10);
  54.         }
  55.         // 3 消费该主题数据
  56.         while (true) {
  57.             ConsumerRecords<String, String> consumerRecords =
  58.                     kafkaConsumer.poll(Duration.ofSeconds(1));
  59.             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
  60.                 System.out.println(consumerRecord);
  61.             }
  62.         }
  63.     }
  64.  
  65. }

 结果：

可以看到都是从0,1分区中，offset为10的位置开始查询的。 

注意：每次执行完，需要修改消费者组名；

2.4  指定offset设置为earliest

auto.offset.reset = earliest | latest | none   其中默认是 latest。本案例设置为earliest。

注意：每次执行完，需要修改消费者组名；每次执行要起一个不同的消费组的名字

代码

 

  1. package com.ljf.spring.boot.demo.consumer;
  2.  
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.common.TopicPartition;
  8. import org.apache.kafka.common.serialization.StringDeserializer;
  9.  
  10. import java.time.Duration;
  11. import java.util.ArrayList;
  12. import java.util.Properties;
  13.  
  14. /**
  15.  * @ClassName: ConsumerDefineOffset
  16.  * @Description: TODO
  17.  * @Author: liujianfu
  18.  * @Date: 2022/04/10 16:30:01
  19.  * @Version: V1.0
  20.  **/
  21. public class ConsumerDefineOffset {
  22.     public static void main(String[] args) {
  23.         // 1.创建消费者的配置对象
  24.         Properties properties = new Properties();
  25.         // 2.给消费者配置对象添加参数
  26.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  27.                 "192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");
  28.         // 配置序列化 必须
  29.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  30.                 StringDeserializer.class.getName());
  31.  
  32.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  33.                 StringDeserializer.class.getName());
  34.         //设置读取的offset的位置
  35.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
  36.         //properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  37.         //properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
  38.         // 配置消费者组（必须），名字可以任意起
  39.         properties.put(ConsumerConfig.GROUP_ID_CONFIG,"shanghai");//注意：每次执行完，需要修改消费者组名；
  40.         KafkaConsumer<String, String> kafkaConsumer = new
  41.                 KafkaConsumer<>(properties);
  42.         // 消费某个主题的某个分区数据，0号分区
  43.         ArrayList<TopicPartition> topicPartitions = new
  44.                 ArrayList<>();
  45.         topicPartitions.add(new TopicPartition("kafka-ljf", 0));
  46.         kafkaConsumer.assign(topicPartitions);
  47.         while (true){
  48.             ConsumerRecords<String, String> consumerRecords =
  49.                     kafkaConsumer.poll(Duration.ofSeconds(1));
  50.             for (ConsumerRecord<String, String> consumerRecord :
  51.                     consumerRecords) {
  52.                 System.out.println(consumerRecord);
  53.             }
  54.         }
  55.     }
  56. }

3.执行结果

 2.5  指定时间消费数据

在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。
例如要求按照时间消费前一天的数据，怎么处理？ 
  1. package com.ljf.spring.boot.demo.consumer;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5. import org.apache.kafka.common.serialization.StringDeserializer;
  6.  
  7. import java.time.Duration;
  8. import java.util.*;
  9.  
  10. /**
  11.  * @ClassName: ConsumerRangeTime
  12.  * @Description: TODO
  13.  * @Author: liujianfu
  14.  * @Date: 2022/04/10 16:53:36
  15.  * @Version: V1.0
  16.  **/
  17. public class ConsumerRangeTime {
  18.     public static void main(String[] args) {
  19.         Properties properties = new Properties();
  20.         // 连接
  21.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  22.                 "192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");
  23.         // key value 反序列化
  24.  
  25.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  26.                 StringDeserializer.class.getName());
  27.  
  28.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  29.                 StringDeserializer.class.getName());
  30.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-time");
  31.         // 1 创建一个消费者
  32.         KafkaConsumer<String, String> kafkaConsumer = new
  33.                 KafkaConsumer<>(properties);
  34.         // 2 订阅一个主题
  35.         ArrayList<String> topics = new ArrayList<>();
  36.         topics.add("kafka-ljf");
  37.         kafkaConsumer.subscribe(topics);
  38.         Set<TopicPartition> assignment = new HashSet<>();
  39.         while (assignment.size() == 0) {
  40.             kafkaConsumer.poll(Duration.ofSeconds(1));
  41.             // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
  42.             assignment = kafkaConsumer.assignment();
  43.         }
  44.         HashMap<TopicPartition, Long> timestampToSearch = new
  45.                 HashMap<>();
  46.         // 封装集合存储，每个分区对应一天前的数据
  47.         for (TopicPartition topicPartition : assignment) {
  48.             timestampToSearch.put(topicPartition, System.currentTimeMillis() - 5 * 24 * 3600 * 1000);
  49.         }
  50.         // 获取从 1 天前开始消费的每个分区的 offset
  51.         Map<TopicPartition, OffsetAndTimestamp> offsets =
  52.                 kafkaConsumer.offsetsForTimes(timestampToSearch);
  53.         // 遍历每个分区，对每个分区设置消费时间。
  54.         for (TopicPartition topicPartition : assignment) {
  55.             OffsetAndTimestamp offsetAndTimestamp =
  56.                     offsets.get(topicPartition);
  57.             // 根据时间指定开始消费的位置
  58.             if (offsetAndTimestamp != null){
  59.                 kafkaConsumer.seek(topicPartition,
  60.                         offsetAndTimestamp.offset());
  61.             }
  62.         }
  63.         // 3 消费该主题数据
  64.         while (true) {
  65.             ConsumerRecords<String, String> consumerRecords =
  66.                     kafkaConsumer.poll(Duration.ofSeconds(1));
  67.             for (ConsumerRecord<String, String> consumerRecord :
  68.                     consumerRecords) {
  69.                 System.out.println(consumerRecord);
  70.             }
  71.         }
  72.     }
  73.  
  74. }

结果：

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/638944
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号