【Kafka-3.x-教程】-【四】Kafka-消费者-Consumer

【Kafka-3.x-教程】专栏：

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门
【Kafka-3.x-教程】-【二】Kafka-生产者-Producer
【Kafka-3.x-教程】-【三】Kafka-Broker、Kafka-Kraft
【Kafka-3.x-教程】-【四】Kafka-消费者-Consumer
【Kafka-3.x-教程】-【五】Kafka-监控-Eagle
【Kafka-3.x-教程】-【六】Kafka 外部系统集成 【Flume、Flink、SpringBoot、Spark】
【Kafka-3.x-教程】-【七】Kafka 生产调优、Kafka 压力测试

【Kafka-3.x-教程】-【四】Kafka-消费者-Consumer

  • 1）Kafka 消费方式
  • 2）Kafka 消费者工作流程
    • 2.1.消费者总体工作流程
    • 2.2.消费者组原理
    • 2.3.消费者组初始化流程
    • 2.4.消费者组详细消费流程
    • 2.5.消费者重要参数
  • 3）消费者 API
    • 3.1.独立消费者案例（订阅主题）
    • 3.2.独立消费者案例（订阅分区）
    • 3.3.消费者组案例
  • 4）分区的分配以及再平衡
    • 4.1.Range 以及再平衡
    • 4.2.RoundRobin 以及再平衡
    • 4.3.Sticky 以及再平衡
  • 5）offset 位移
    • 5.1.offset 的默认维护位置
    • 5.2.自动提交 offset
    • 5.3.手动提交 offset
    • 5.4.指定 Offset 消费
    • 5.5.指定时间消费
    • 5.6.漏消费和重复消费
  • 6）消费者事务
  • 7）数据积压

1）Kafka 消费方式

在这里插入图片描述

2）Kafka 消费者工作流程

2.1.消费者总体工作流程

在这里插入图片描述

1、一个消费者可以消费多个分区的数据。消费者之间是完全独立的。

2、每个分区的数据只能由消费者组中的一个消费者进行消费。

3、Kafka 的 offset（偏移量）负责记录消费者或消费者组消费数据的位置。

4、新版本的 Kafka 由 __consumer_offsets 这个主题来记录 offset。老版本是存储在 ZK 的 consumer 目录下，会产生消费者和 ZK 的大量频繁的交互，网络压力较大。

5、Kafka 的 offset 是持久化到硬盘中的，所以数据安全性得以保障。

2.2.消费者组原理

在这里插入图片描述

在这里插入图片描述

1、一个消费者组中包含多个消费者，形成一个消费者组的条件是，消费者对应的 group.id 一致。

（1）消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者进行消费。

（2）消费者组之间互不影响，可以把消费者组理解成为一个包装后的消费者。

2、如果消费者组中的消费者数量大于分区数量，那么消费者组中必定会出现闲置的消费者。

2.3.消费者组初始化流程

在这里插入图片描述

消费者是如何形成消费者组的呢？

1、coordinator：辅助实现消费者组的初始化和分区的分配。

（1）每个 Broker 中都会存在一个 coordinator。

（2）coordinator 的选择取决于 group.id 的 hashcode % 50（__consumer_offsets 这个主题的分区数），假如取得的值为 1，那么就找到 1 号分区所在的节点进行连接。

2、每个 Consumer 都发送 join Group 的请求。

3、从众多 Consumer 中随机选择一个 Consumer - Leader。

4、coordinator 把要消费的 topic 详情发送给 Consumer - Leader。

5、Consumer - Leader 制定消费方案。

6、Consumer - Leader 将制定计划发送给 coordinator。

7、coordinator 将对应计划发布到各个 Consumer。

注意：每个消费者都会定时和 coordinator 保持心跳（默认 3 秒），一旦超时（session.timeout.ms=45s），coordinator 会认为消费者挂了，该消费者会被移除，并触发再平衡；消费者处理的时间过长（max.poll.interval.ms 5分钟），也会将这个消费者下线，并触发再平衡。

2.4.消费者组详细消费流程

在这里插入图片描述

1、消费者组创建网络连接客户端（ConsumerNetworkClient），用于和集群进行交互。

2、调用 sendFetches（发送消费数据请求）

  • Fetch.min.bytes：每批次最小抓取大小，默认1字节。
  • fetch.max.wait.ms：一批数据最小值未达到的超时时间，默认500ms。
  • Fetch.max.bytes：每批次最大抓取大小，默认50m。

3、ConsumerNetworkClient 发送 send 请求，Kafka 集群通过回调方法（onSuccess），把对应的结果拉取过来。

4、拉取到的数据会放到 completedFetches 的消息队列中（queue）。

5、消费者从队列中拉取数据，默认一次 500 条（Max.poll.records）。

（1）反序列化操作.

（2）拦截器操作（用于处理数据）。

（3）处理数据。

2.5.消费者重要参数

在这里插入图片描述在这里插入图片描述

3）消费者 API

3.1.独立消费者案例（订阅主题）

1、需求：创建一个独立消费者，消费 first 主题中数据。

在这里插入图片描述

注意：在消费者 API 代码中必须配置消费者组 id。命令行启动消费者不填写消费者组id 会被自动填写随机的消费者组 id。

2、编写代码：

public class CustomConsumer {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test5");// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}kafkaConsumer.commitAsync();}}
}

3、测试

（1）在 IDEA 中执行消费者程序。

（2）在 Kafka 集群控制台，创建 Kafka 生产者，并输入数据。

bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first>hello

（3）在 IDEA 控制台观察接收到的数据。

ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3,  offset= 0, CreateTime = 1629160841112, serialized key size = -1,  serialized value size = 5, headers = RecordHeaders(headers = [],  isReadOnly =
false), key = null, value = hello)

3.2.独立消费者案例（订阅分区）

1、需求：创建一个独立消费者，消费 first 主题 0 号分区的数据。

在这里插入图片描述

2、代码编写：

public class CustomConsumerPartition {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 1 创建一个消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题对应的分区ArrayList<TopicPartition> topicPartitions = new ArrayList<>();topicPartitions.add(new TopicPartition("first",0));kafkaConsumer.assign(topicPartitions);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

3、测试
（1）在 IDEA 中执行消费者程序。

（2）在 IDEA 中执行生产者程序 CustomProducerCallback()在控制台观察生成几个 0 号分区的数据。

first 0 381
first 0 382
first 2 168
first 1 165
first 1 166

（3）在 IDEA 控制台，观察接收到的数据，只能消费到 0 号分区数据表示正确。

ConsumerRecord(topic = first, partition = 0, leaderEpoch = 14, 
offset = 381, CreateTime = 1636791331386, serialized key size = -
1, serialized value size = 9, headers = RecordHeaders(headers = 
[], isReadOnly = false), key = null, value = atguigu 0)
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 14, 
offset = 382, CreateTime = 1636791331397, serialized key size = -
1, serialized value size = 9, headers = RecordHeaders(headers = 
[], isReadOnly = false), key = null, value = atguigu 1)

3.3.消费者组案例

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

在这里插入图片描述

2、代码实现：

复制代码为三份，同时启动，模拟三个消费者都在一个消费者组中。

public class CustomConsumer1 {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test5");// 设置分区分配策略properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

3、测试

（1）启动代码中的生产者发送消息，在 IDEA 控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 Thread.sleep(2)）。

ConsumerRecord(topic = first, partition = 0, leaderEpoch = 2, 
offset = 3, CreateTime = 1629169606820, serialized key size = -1, 
serialized value size = 8, headers = RecordHeaders(headers = [], 
isReadOnly = false), key = null, value = hello1)
ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3, 
offset = 2, CreateTime = 1629169609524, serialized key size = -1, 
serialized value size = 6, headers = RecordHeaders(headers = [], 
isReadOnly = false), key = null, value = hello2)
ConsumerRecord(topic = first, partition = 2, leaderEpoch = 3, 
offset = 21, CreateTime = 1629169611884, serialized key size = -1, 
serialized value size = 6, headers = RecordHeaders(headers = [], 
isReadOnly = false), key = null, value = hello3)

（2）观察三份代码的控制台，每份代码中只会消费一个分区中的数据。

（3）重新发送到一个全新的主题中，由于默认创建的主题分区数为 1，可以看到只能有一个消费者消费到数据。

在这里插入图片描述

4）分区的分配以及再平衡

在这里插入图片描述

1、一个 consumer group 中有多个 consumer 组成，一个 topic 有多个 partition 组成，现在的问题是，到底由哪个 consumer 来消费哪个 partition 的数据。

2、Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。

可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是 Range + CooperativeSticky。Kafka 可以同时使用多个分区分配策略。

4.1.Range 以及再平衡

特点：针对一个分区做排序后计算。

在这里插入图片描述
Range 分区分配再平衡案例

1、停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。

  • 1 号消费者：消费到 3、4 号分区数据。

  • 2 号消费者：消费到 5、6 号分区数据。

  • 0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。

说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。

2、再次重新发送消息观看结果（45s 以后）。

  • 1 号消费者：消费到 0、1、2、3 号分区数据。

  • 2 号消费者：消费到 4、5、6 号分区数据。

说明：消费者 0 已经被踢出消费者组，所以重新按照 range 方式分配。

4.2.RoundRobin 以及再平衡

特点：针对所有分区做排序后轮询。

在这里插入图片描述

RoundRobin 分区分配再平衡案例

1、停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。

  • 1 号消费者：消费到 2、5 号分区数据

  • 2 号消费者：消费到 4、1 号分区数据

  • 0 号消费者的任务会按照 RoundRobin 的方式，把数据轮询分成 0 、6 和 3 号分区数据，分别由 1 号消费者或者 2 号消费者消费。

说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。

2、再次重新发送消息观看结果（45s 以后）。

  • 1 号消费者：消费到 0、2、4、6 号分区数据

  • 2 号消费者：消费到 1、3、5 号分区数据

说明：消费者 0 已经被踢出消费者组，所以重新按照 RoundRobin 方式分配。

4.3.Sticky 以及再平衡

特点：尽量均匀随机的分配。

粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。

粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

Sticky 分区分配再平衡案例

1、停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。

  • 1 号消费者：消费到 2、5、3 号分区数据。

  • 2 号消费者：消费到 4、6 号分区数据。

  • 0 号消费者的任务会按照粘性规则，尽可能均衡的随机分成 0 和 1 号分区数据，分别由 1 号消费者或者 2 号消费者消费。

说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。

2、再次重新发送消息观看结果（45s 以后）。

  • 1 号消费者：消费到 2、3、5 号分区数据。

  • 2 号消费者：消费到 0、1、4、6 号分区数据。

说明：消费者 0 已经被踢出消费者组，所以重新按照粘性方式分配。

5）offset 位移

5.1.offset 的默认维护位置

在这里插入图片描述

1、__consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.id+topic+分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 进行 compact，也就是每个 group.id+topic+分区号就保留最新数据。

2、消费 offset 案例

思想：__consumer_offsets 为 Kafka 中的 topic，那就可以通过消费者进行消费。

（1）在配置文件 config/consumer.properties 中添加配置 exclude.internal.topics=false，默认是 true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为 false。

（2）采用命令行方式，创建一个新的 topic。

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2

（3）启动生产者往 atguigu 生产数据。

bin/kafka-console-producer.sh --topic atguigu --bootstrap-server hadoop102:9092

（4）启动消费者消费 atguigu 数据。

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic atguigu --group test

注意：指定消费者组名称，更好观察数据存储位置（key 是 group.id+topic+分区号）。

（5）查看消费者消费主题__consumer_offsets。

bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

[offset,atguigu,1]::OffsetAndMetadata(offset=7, 
leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, 
expireTimestamp=None)
[offset,atguigu,0]::OffsetAndMetadata(offset=8, 
leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, 
expireTimestamp=None)

5.2.自动提交 offset

在这里插入图片描述

在这里插入图片描述
1、原理：

（1）消费者主动去拉取数据。

（2）到达 5s 时 Consumer 自动提交 offset。

2、代码示例：

public class CustomConsumerAutoOffset {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 提交时间间隔properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

5.3.手动提交 offset

在这里插入图片描述

1、原理：

  • commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。

  • commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了。

2、代码示例：

（1）由于同步提交 offset 有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交 offset 的示例。

public class CustomConsumerByHandSync {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 手动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}// 手动提交offsetkafkaConsumer.commitSync();}}
}

（2）虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交 offset 的方式。以下为异步提交 offset 的示例：

public class CustomConsumerByHandSync {public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 手动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}// 手动提交offsetkafkaConsumer.commitAsync();}}
}

5.4.指定 Offset 消费

在这里插入图片描述

auto.offset.reset = earliest | latest | none 默认是 latest。

当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？

1、earliest：自动将偏移量重置为最早的偏移量，–from-beginning。

2、latest（默认值）：自动将偏移量重置为最新偏移量。

3、none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

代码示例：

public class CustomConsumerSeek {public static void main(String[] args) {// 0 配置信息Properties properties = new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test3");// 1 创建消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题ArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 指定位置进行消费Set<TopicPartition> assignment = kafkaConsumer.assignment();//  保证分区分配方案已经制定完毕while (assignment.size() == 0){kafkaConsumer.poll(Duration.ofSeconds(1));assignment = kafkaConsumer.assignment();}// 指定消费的offsetfor (TopicPartition topicPartition : assignment) {kafkaConsumer.seek(topicPartition,600);}// 3  消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

5.5.指定时间消费

需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。

例如要求按照时间消费前一天的数据，怎么处理？

public class CustomConsumerSeekTime {public static void main(String[] args) {// 0 配置信息Properties properties = new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test3");// 1 创建消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题ArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 指定位置进行消费Set<TopicPartition> assignment = kafkaConsumer.assignment();//  保证分区分配方案已经制定完毕while (assignment.size() == 0){kafkaConsumer.poll(Duration.ofSeconds(1));assignment = kafkaConsumer.assignment();}// 希望把时间转换为对应的offsetHashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();// 封装对应集合for (TopicPartition topicPartition : assignment) {topicPartitionLongHashMap.put(topicPartition,System.currentTimeMillis() - 1 * 24 * 3600 * 1000);}Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);// 指定消费的offsetfor (TopicPartition topicPartition : assignment) {OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());}// 3  消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

5.6.漏消费和重复消费

重复消费：已经消费了数据，但是 offset 没提交。

漏消费：先提交 offset 后消费，有可能会造成数据的漏消费。

在这里插入图片描述

思考：怎么能做到既不漏消费也不重复消费呢？详看消费者事务。

6）消费者事务

注意：下游必须支持事务才能做到精准一次消费。

在这里插入图片描述

7）数据积压

1、增大分区与消费者核数（消费者数 = 分区数）

2、增大每次拉取的吞吐量。

在这里插入图片描述

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/613669.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

用握力器玩谷歌小恐龙游戏（二）

用握力器玩谷歌小恐龙游戏（二）

往期回顾 用握力器玩谷歌小恐龙游戏&#xff08;一&#xff09; GS-GAME-PC 前言 专门买了一个电钻&#xff0c;在握力器上钻出了两个孔&#xff0c;用来放两个引脚的按键&#xff0c;这样比之前用热熔胶的方式更干净&#xff0c;更稳定 加了一个18650的电池&#xff0c;之前…
阅读更多...
ChatGPT提示词的高级技巧——解锁AI交互新境界

ChatGPT提示词的高级技巧——解锁AI交互新境界

大家好我是在看&#xff0c;记录普通人学习探索AI之路。 这一讲我来介绍几种提示词的高级使用技巧。 1.拆解问题 对于一个一次性很难完整回答的大问题&#xff0c;我们可以将其拆解成若干子问题&#xff0c;再将这些问题逐步输入。 我们来看下面的例子&#xff1a; 2.迭代改…
阅读更多...
RLHF与LLM训练的碰撞：寻找最佳实践之路！

RLHF与LLM训练的碰撞：寻找最佳实践之路！

了解更多公众号&#xff1a;芝士AI吃鱼 在讨论大型语言模型&#xff08;LLM&#xff09;时&#xff0c;无论是在研究新闻还是教程中&#xff0c;经常提到一个称为“带有人类反馈的强化学习”&#xff08;RLHF&#xff09;的过程。由于RLHF能够将人类偏好纳入优化过程&#xff0…
阅读更多...
What does “grep -i“ do?

What does “grep -i“ do?

grep&#xff08; Global Regular Expression Print &#xff09;根据 给定的正则表达式 搜索文本&#xff0c;并将匹配的行打印出来 grep -i 表示查找的过程中忽略大小写 在已安装的rpm包里&#xff0c;查询是否有tree相关的包 rpm -qa | grep -i "tree"在文件中搜…
阅读更多...
使用Python打造一个爱奇艺热播好剧提前搜系统

使用Python打造一个爱奇艺热播好剧提前搜系统

目录 一、系统功能设计 二、数据获取与处理 三、搜索功能实现 四、用户界面设计 五、系统部署与维护 六、总结 随着互联网的普及和人们对于娱乐需求的增加&#xff0c;视频网站成为了人们观看电视剧、电影等视频内容的主要渠道。爱奇艺作为国内知名的视频网站之一&#x…
阅读更多...
使用 Asp.net core webapi 集成配置系统，提高程序的灵活和可维护性

使用 Asp.net core webapi 集成配置系统，提高程序的灵活和可维护性

前言&#xff1a;什么是集成配置系统&#xff1f; 集成配置系统的主要目的是将应用程序的配置信息与代码分离&#xff0c;使得配置信息可以在不需要修改代码的情况下进行更改。这样可以提高应用程序的灵活性和可维护性。 ASP.NET Core 提供了一种灵活的配置系统&#xff0c;可…
阅读更多...
Flink构造宽表实时入库案例介绍

Flink构造宽表实时入库案例介绍

1. 安装包准备 Flink 1.15.4 安装包 Flink cdc的mysql连接器 Flink sql的sdb连接器 MySQL驱动 SDB驱动 Flink jdbc的mysql连接器 2. 入库流程图 3. Flink安装部署 上传Flink压缩包到服务器&#xff0c;并解压 tar -zxvf flink-1.14.5-bin-scala_2.11.tgz -C /opt/ 复…
阅读更多...
显示器新赛道Type-C接口

显示器新赛道Type-C接口

如果把主机比作大脑&#xff0c;那显示器就是眼睛&#xff0c;没有眼睛&#xff0c;大脑再强大也发挥不出效果&#xff0c;所以显示器作为电脑最重要的输出设备&#xff0c;有着举足轻重的地位&#xff0c;可以说在生活中处处都有显示器的影子。其实显示器的历史也是科技发展史…
阅读更多...
涛思数据获评北京市“专精特新”中小企业

涛思数据获评北京市“专精特新”中小企业

众所周知&#xff0c;“专精特新”企业是国家引导中小企业增强自主创新能力和核心竞争力&#xff0c;不断提高中小企业发展质量和水平而实施的重大工程&#xff0c;旨在支持企业走专精特新发展之路&#xff0c;更好地促进企业高质量发展&#xff0c;也成为各领域产业链供应的关…
阅读更多...
YOLOv8 Ultralytics：使用Ultralytics框架进行定向边界框对象检测

YOLOv8 Ultralytics：使用Ultralytics框架进行定向边界框对象检测

YOLOv8 Ultralytics&#xff1a;使用Ultralytics框架进行定向边界框对象检测 前言相关介绍前提条件实验环境安装环境项目地址LinuxWindows 使用Ultralytics框架进行定向边界框对象检测参考文献 前言 由于本人水平有限&#xff0c;难免出现错漏&#xff0c;敬请批评改正。更多精…
阅读更多...
Linux 基于 rsync 实现集群分发脚本 xsync

Linux 基于 rsync 实现集群分发脚本 xsync

一、rsync 简介 rsync&#xff08;remote synchronize&#xff09;是 Liunx/Unix 下的一个远程数据同步工具。它可以通过 LAN/WAN 快速同步多台主机间的文件和目录&#xff0c;并适当利用 rsync 算法&#xff08;差分编码&#xff09;以减少数据的传输。 rsync 算法并不是每一次…
阅读更多...
QT问题 ui提升部件时No such file or directory

QT问题 ui提升部件时No such file or directory

问题: qt使用ui对部件提升在编译时找不到对应的头文件 出错原因: 因为将部件提升为自定义部件后&#xff0c;在编译时会去默认的路径下去找头文件&#xff0c;而自定义的头文件并不在默认路径文件下&#xff0c;而是在当前目录下&#xff0c;所以这个时候需要自己指定出自…
阅读更多...
回归预测 | Matlab实现RIME-HKELM霜冰算法优化混合核极限学习机多变量回归预测

回归预测 | Matlab实现RIME-HKELM霜冰算法优化混合核极限学习机多变量回归预测

回归预测 | Matlab实现RIME-HKELM霜冰算法优化混合核极限学习机多变量回归预测 目录 回归预测 | Matlab实现RIME-HKELM霜冰算法优化混合核极限学习机多变量回归预测效果一览基本介绍程序设计参考资料 效果一览 基本介绍 1.Matlab实现RIME-HKELM霜冰算法优化混合核极限学习机多变…
阅读更多...
Redis 发布订阅

Redis 发布订阅

目录 1.Redis Unsubscribe 命令 - 指退订给定的频道。简介语法可用版本: > 2.0.0返回值: 这个命令在不同的客户端中有不同的表现。 示例 2.Redis Subscribe 命令 - 订阅给定的一个或多个频道的信息。简介语法可用版本: > 2.0.0返回值: 接收到的信息 示例 3.Redis Pubsub …
阅读更多...
Camunda Spin

Camunda Spin

Spin 常用于在脚本中解析json或者xml使用&#xff0c;S(variable) 表示构造成Spin对象&#xff0c;通过prop(“属性名”)获取属性值&#xff0c;通过stringValue()、numberValue()、boolValue() 等对类型转换。 repositoryService.createDeployment().name("消息事件流程&…
阅读更多...
web第一次作业

web第一次作业

题1&#xff1a; <form action"#" method"post"><table><tr><td>用户名&#xff1a;</td><td><input type"text" name"UserName" maxlength"20" size"15"></td>…
阅读更多...
Javascript——vue下载blob文档流

Javascript——vue下载blob文档流

<el-table-column label"操作" fixed"right" width"150" showOverflowTooltip><template slot-scope"scope"><el-button type"text" v-has"stbsd-gjcx-down" class"edit-button" click&…
阅读更多...
MySQL 从零开始：02 MySQL 安装

MySQL 从零开始：02 MySQL 安装

文章目录 1、下载 MySQL 安装程序2、安装 MySQL 要操作 MySQL &#xff0c;首先要安装 MySQL &#xff0c;本文将一步步展示如何安装 MySQL&#xff0c;简直详细到令人发指。 环境&#xff1a; 操作系统&#xff1a;Windows10 64位MySQL版本&#xff1a;社区版 8.0.11.0 1、下…
阅读更多...
探索AI技术的奥秘：揭秘人工智能的核心原理

探索AI技术的奥秘：揭秘人工智能的核心原理

人工智能&#xff08;AI&#xff09;已经成为当今科技领域最引人注目的话题之一。然而&#xff0c;对于许多人来说&#xff0c;AI仍然是一个神秘的领域&#xff0c;不清楚其核心原理。本文将探索AI技术的奥秘&#xff0c;为读者揭开人工智能的核心原理。 随着技术的飞速发展&am…
阅读更多...
Asynchronous FIFO and synchronous FIFO-翻译自外网

Asynchronous FIFO and synchronous FIFO-翻译自外网

Synchronous FIFO 先进先出 (FIFO) 是一种非常流行且有用的设计块&#xff0c;用于模块之间的同步和握手机制。 FIFO 的深度&#xff1a; FIFO 中的槽数或行数称为 FIFO 的深度。 FIFO 的宽度&#xff1a;每个槽或行中可以存储的位数称为 FIFO 的宽度。 在同步 FIFO 中&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023