首页 > 编程入门 文章详情

4.kafka3.x消费者相关原理

编程入门 行业动态 更新时间:2024-10-15 00:26:55

4.kafka3.x消费者相关<a href=https://www.elefans.com/category/jswz/34/1770123.html style=原理"/>

4.kafka3.x消费者相关原理

kafka消费方式

  1. pull 拉模式 consumer从broker中主动拉取数据,kafka采用这种模式
  2. push模式 由broker主动推送消息consumer,kafka没有采用这种模式,因为每个consumer的消费速度不通,broker很难把握发送消息的速度
  3. pull模式的不足之处在于如果kafka没有数据,consumer可能会一直请求到空数据

kafka消费流程

  1. producer将消息发送给broker中partition中的leader,并记录offset
  2. 不同的consumer可以消费多个分区的数据,但是每一个partition只能由消费者组中的一个消费者消费
  3. consumer group中的某一个消费者对partition中的消息进行消费
  4. 每个消费者的offset由消费者提交到系统主题(__consumer_offset)进行保存

消费者组

Consumer Group (CG): 消费者组,由多个consumer组成,形成一个CG的前提是多个消费者有相同的groupid

  • 消费者组内每一个消费者负责消费不同分区的数据,一个分区只能由一个组内的消费者消费
  • 消费者组之间互不影响,所有消费者都属于某一个消费者组,即消费者组是一个逻辑的订阅者

消费者组初始化流程

coordinator辅助实现消费者组的初始化和 分区的分配,coordinator节点选择 groupid的hashcode值%50(__consumer_offset的分区数量),例如groupid的hashcode是1 1%50=1, 那么该consumer_offset的对应主题的1号分区所在的broker即使改节点coordinator作为该消费者组的老大,消费者组下所有的消费者提交的offset都会往该分区上去提交offset

  1. 每一个consumer发送joingroup请求
  2. cordinator选出一个consumer作为该group的leader
  3. CG中的所有consumer将消费的topic的情况发送给consumer leader
  4. consumer leader负责指定消费计划并发送给coordinator
  5. coordinator将消费方案下发给各个consumer
  6. 每一个consumer和coordinator保持心跳

消费者参数

参数名称描述
bootstrap.servers向 Kafka 集群建立初始连接用到的 host/port 列表
key.deserializer 和 value.deserializer指定接收消息的 key 和 value 的反序列化类型。一定要写全 类名。
group.id标记消费者所属的消费者组。
enable.automit默认值为 true，消费者会自动周期性地向服务器提交偏移 量
automit.interval.ms如果设置了 enable.automit 的值为 true， 则该值定义了 消费者偏移量向 Kafka 提交的频率，默认 5s
auto.offset.reset当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了），该如何处理？ earliest：自动重置偏 移量到最早的偏移量。 latest：默认，自动重置偏移量为最 新的偏移量。 none：如果消费组原来的（previous）偏移量 不存在，则向消费者抛异常。 anything：向消费者抛异常
offsets.topic.num.partitions__consumer_offsets 的分区数，默认是 50 个分区。
heartbeat.interval.msKafka 消费者和 coordinator 之间的心跳时间，默认 3s。 该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的 1/3
session.timeout.msKafka 消费者和 coordinator 之间连接超时时间，默认 45s。 超过该值，该消费者被移除，消费者组执行再平衡
max.poll.interval.ms消费者处理消息的最大时长，默认是 5 分钟。超过该值，该 消费者被移除，消费者组执行再平衡。
fetch.min.bytes默认 1 个字节。消费者获取服务器端一批消息最小的字节 数
fetch.max.wait.ms默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据
fetch.max.bytes默认 Default: 52428800（50 m）。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响
max.poll.records一次 poll 拉取数据返回消息的最大条数，默认是 500 条

消费者组详细消费流程

代码实现

public class KafkaConsumerDemo {public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");// group.idproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer.cg1");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);Collection<String> topics = new ArrayList<>();// 订阅的主题topics.add("test_topic");consumer.subscribe(topics);while (true) {// 每3秒拉一次消息ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(3));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord.toString());}}}
}

单独订阅分区代码 org.apache.kafkamon.TopicPartition

public class KafkaSpecialConsumerDemo {public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");// group.idproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer.cg1");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);// 独立分区的方式进行消费List<TopicPartition> topics = new ArrayList<>();topics.add(new TopicPartition("test_topic", 0));// 订阅的主题consumer.assign(topics);while (true) {// 每3秒拉一次消息ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(3));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord.toString());}}}
}

分区的分配再平衡

修改分区规则

ArrayList<String> startegys = new ArrayList<>();
startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,startegys);
  1. 一个CG由多个consumer组成,一个topic由多个partition组成,一个topic中的partition只能对应一个CG中的consumer
  2. kafka有四种主流的分区策略: Range RoundBobin Sticky CooperativeSticky,可以通过partition.assignment.strategy进行修改默认策略是Range + CooperativeSticky,kafka可以同时使用多种分区策略
  3. bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic test_topic --partitions 7 可以修改分区, 分区只能增加不能减少
Range分区策略

Range是对每个topic而言

  1. 首先对同一个topic里面的分区按序号进行排序,并对消费者按字母顺序排序 例如7个分区,3个消费者,排序之后分区为0,1,2,3,4,5,6,消费者排序之后为C0,C1,C2
  2. 通过partition/consumer来决定每个消费者消费几个分区,如果无法整除,前面几个消费者会多分配1个分区 如7/3=2 余数为1 C0消费3个分区 C1,C2消费2个分区
  3. 存在的问题: 数据偏斜,单独对一个topic来说 C0只多消费一个分区影响不大,如果存在N多个topic,每一个topic对C0来说都多消费一个partition,当topic越来越多C0的压力就会有越来越大

range分区再平衡实例

  1. 还是以7个分区 3个消费者为例 以range 分区策略 C0(0,1,2) ,C1(3,4),C2(5,6)
  2. 停掉C0,快速重新发送消费45S之内
    1. C1消费3,4号分区
    2. C2消费5,6号分区
    3. C0消费任务会整体分配到C1和C2
    4. C0挂掉之后,消费者组需要根据超时时间45S判断他是否退出,45S之后就会将他踢出CG,把他的任务分配到其他broker
  3. 45S之后
    1. C1消费0,1,2,3分区数据
    2. C2消费4,5,6分区数据
    3. 说明C0已经被踢出CG,触发了range的再平衡
RoundRobin与再平衡

RoundRobin将所有partition和consumer都列出来,按照hashcode进行排序,再通过轮询算法来分配partition到各个消费者

7个分区0,1,2,3,4,5,6 3个consumer

0->C1

1->C2

2->C3

3->C1 依次类推

再平衡

  1. C1宕机之后,C0按照RoundRobin方式将数据轮训分给0,3,6分区数据分别由C2和C3消费
  2. 45S之后 C2,C3根据RoundRobin再平衡变成C2(0,2,4,6) C3(1,3,5)
Sticky与再平衡

粘性分区:在执行新的分配之前先考虑上次分配的结果,尽量少的调整分配的变动,可以节省大量开销

kafka0.11x引入这种策略,首先尽量均衡的放置分区在消费者上,在出现同一个CG中的consumer出现问题的时候,会尽量保持原有分配的分区不变

粘性分区跟加注重分区的均衡性，随机将分区均匀的分配到各个consumer中

offset

kafka0.9之前offset是维护在ZK中,维护在ZK中存在的问题是consumer必须和ZK保持长时间的交互导致当topic增多之后会增加ZK和consumer的负担

0.9之后kafka将offset维护在了系统主题__consumer_offsets中,该主题有50个partition,采用K-V方式存储数据,key=groupId+topic+partition号,value即使当前的offset值.每隔一段时间,kafka内部会对这个topic进行压缩compact操作,保留最新的offset

  1. __consumer_offset是kafka的topic,也可以通过消费者消费
  2. 默认consumer不能消费系统主题,可以通过config/consumer.properties添加配置exclude.internal.topics默认为true,改为false即可消费系统主题

实验

  1. 创建一个topic

    [root@node1 bin]# bash kafka-topics.sh --bootstrap-server node1:9092 --topic testOffset --partitions 2 --create 
Created topic testOffset.

  2. 生产数据

    bash kafka-console-producer.sh --bootstrap-server node1:9092 --topic testOffset

  3. 启动消费者消费该主题 为方便查看 添加–group参数增加group.id

    bash kafka-console-consumer.sh --bootstrap-server node1:9092 --topic testOffset --from-beginning --group test

  4. 查看消费者系统主题__consumer_offsets

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

自动提交offset

kafka提供了自动提交offset功能

参数名称描述
enable.automit默认值为 true，消费者会自动周期性地向服务器提交偏移量
automit.interval.ms如果设置了 enable.automit 的值为 true， 则该值定义了消 费者偏移量向 Kafka 提交的频率，默认 5
 // 是否自动提交 offsetproperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
true);// 提交 offset 的时间周期 1000ms，默认 5s
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
1000);

手动提交offset

自动offset提交是基于时间的,开发人员难以把我offset的提交时机,kafka提供了手动提交offset的API

手动提交的方式有两种,分别是commitSync(同步提交) 和 commitAsync(异步提交).两者都会将本次提交的一批数据的最高的偏移量提交,区别在于同步提交会阻塞当前线程,直到提交成功,并且会自动失败重试.异步提交没有重试机制可能会提交失败

  • commitSync: 必须等到offset提交完毕,再消费下一批数据
  • commitAsync:发送完offset之后,就开始消费下一批数据
public class ManualCommitConsumerDemo {public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");// group.idproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer.cg1");properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);Collection<String> topics = new ArrayList<>();// 订阅的主题topics.add("test_topic");consumer.subscribe(topics);while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> record : records) {System.out.println(record);}// 同步提交offset
//            consumermitSync();// 异步提交offsetconsumermitAsync();}}
}

指定offset消费

auto.offset.reset

  • earliest: 自动将偏移量重置为最早的偏移量，类似 --from-beginning
  • latest (默认): 自动将偏移量重置为最新偏移量
  • none: 如果未找到消费者组的先前偏移量则向消费者抛出异常
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

指定任意位置进行消费

org.apache.kafka.clients.consumer.Consumer#seek(org.apache.kafkamon.TopicPartition, long)

public class KafkaSeekOffsetConsumerDemo {public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");// group.idproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer.haha");properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);Collection<String> topics = new ArrayList<>();// 订阅的主题topics.add("test_topic");consumer.subscribe(topics);// 为了避免程序执行到这里kafka链接还没建立完成Set<TopicPartition> topicPartitions = new HashSet<>();while (topicPartitions.size() <= 0) {consumer.poll(Duration.ofSeconds(1));// 获取所有的topic和partition信息topicPartitions = consumer.assignment();}// 遍历所有分区，并指定 offset 从 5 的位置开始消费for (TopicPartition topicPartition : topicPartitions) {consumer.seek(topicPartition, 5);}while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));for (ConsumerRecord<String, String> record : records) {System.out.println(record);}}}
}

指定时间点开始消费

生产环境中可能会遇到一段时间的消费数据异常需要重新消费,此时kafka可以变相的根据时间点来消费

kafka没有提供直接通过时间来消费的API,但是提供了通过时间来计算出offset的API

KafkaConsumer#offsetsForTimes(java.util.Map<org.apache.kafkamon.TopicPartition,java.lang.Long>)

public class KafkaSeekOffsetTimeConsumer {public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");// group.idproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer.hrhr");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);Collection<String> topics = new ArrayList<>();// 订阅的主题topics.add("test_topic");consumer.subscribe(topics);// 为了避免程序执行到这里kafka链接还没建立完成Set<TopicPartition> topicPartitions = new HashSet<>();while (topicPartitions.size() <= 0) {consumer.poll(Duration.ofSeconds(1));// 获取所有的topic和partition信息topicPartitions = consumer.assignment();}// 封装集合存储，每个分区对应一天前的数据Map<TopicPartition, Long> searchByTimePartition = new HashMap<>();for (TopicPartition topicPartition : topicPartitions) {// 封装集合存储，每个分区对应一天前的数据searchByTimePartition.put(topicPartition, System.currentTimeMillis() - 60 * 60 * 24 * 1000);}//获取从 1 天前开始消费的每个分区的 offsetMap<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = consumer.offsetsForTimes(searchByTimePartition);// 遍历分区 seek到具体的offsetfor (TopicPartition topicPartition : topicPartitions) {if (topicPartitionOffsetAndTimestampMap.get(topicPartition) != null) {consumer.seek(topicPartition, topicPartitionOffsetAndTimestampMap.get(topicPartition).offset());}}while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));for (ConsumerRecord<String, String> record : records) {System.out.println(record);}}}
}

漏消息和重复消费

  • 重复消费: 消费了数据没有提交offset

    比如开启了自动offset提交,consumer默认5s提交一次offset,提交offset 2s之后consumer挂了,此时已经消费了2s的消息,但是因为没有触发5s时间间隔没有告诉kafka已经消费信息,此时再启动consumer broker还是记录的5s自动提交之前的offset 此时会造成消息的重复消费

  • 漏消费: 提交了offset后消费,可能会造成数据漏消费

    如果将offset设置为手动提交,当offset被提交时,数据还在内存中位落盘,刚好消费者宕机,offset已经提交,数据未处理,此时就算再启动consumer也消费不到之前的数据了,导致了数据漏消费

如果想要consumer精准一次消费,需要kafka消息的消费过程和提交offset变成原子操作.此时需要我们将kafka的offset持久化到其他支持事务的中间件(比如MySQL)

消息积压

  1. 如果是kafka消费能力不足,考虑增加topic分区数,并且同时增加消费者组的消费者数量.因为一个partition只能被CG中的一个consumer消费,所以partition和consumer必须同时增加
  2. 如果是下游数据处理不及时,可以提高每次拉取的数量.批次拉取数据过少,使得处理数据小于生产的数据
参数名称描述
fetch.max.bytes默认 Default: 52428800（50 m）。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝 对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records一次 poll 拉取数据返回消息的最大条数，默认是 500 条

更多推荐

4.kafka3.x消费者相关原理

本文发布于:2024-03-12 03:54:30，感谢您对本站的认可！
本文链接:https://www.elefans.com/category/jswz/34/1730635.html
版权声明:本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系，我们将在24小时内删除。
本文标签：原理   消费者

发布评论

评论列表 （有 0 条评论）

最近发表

草根站长

>www.elefans.com

编程频道|电子爱好者 - 技术资讯及电子产品介绍！

热门文章

标签列表