1、消费者
1.1 Kafka消费方式
1、pull(拉)模式:consumer采用从broker中主动拉取数据。
2、push(推)模式:Kafka没有采用这种方式。因为broker决定消息发生速率,很难适应所有消费者的消费速率。例如推送的速度是50M/s,Consumer1、Consumer2就来不及处理消息。
pull模式不足之处是如果Kafka没有数据,消费者可能会陷入循环中,一直返回空数据。
1.2 Kafka消费者工作流程
1.2.1 消费者总体工作流程
1.2.2 消费者组原理
Consumer Group(CG):消费者组,由多个consumer组成。形成一个消费者组的条件,是所有消费者的groupid相同。
- 消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费。
- 消费者组之间互不影响。所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者。
1、消费者组初始化流程
(1)coordinator:辅助实现消费者组的初始化和分区的分配
coordinator节点选择=groupid的hashcode值%50(_consumer_offsets的分区数量)
例如:groupid的hashcode值=1,1%50=1,那么_consumer_offsets主题的1号分区,在哪个broker上,就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。
2、消费者组详细消费流程
1.2.3 消费者重要参数
参数名称描述bootstrap.servers向 Kafka 集群建立初始连接用到的 host/port 列表key.deserializer 和value.deserializer指定接收消息的 key 和 value 的反序列化类型。一定要写全类名group.id标记消费者所属的消费者组。enable.auto.commit默认值为 true,消费者会自动周期性地向服务器提交偏移量。auto.commit.interval.ms如果设置了 enable.auto.commit 的值为 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 条。
1.3 消费者API
1.3.1 独立消费者案例(订阅主题)
1、需求
创建一个独立的消费者,消费first主题中的数据
注意:在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id。
2、实现步骤
(1)创建包名:com.zhm.consumer
(2)编写代码
packageorg.zhm.consumer;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.common.serialization.StringDeserializer;importjava.time.Duration;importjava.util.ArrayList;importjava.util.Properties;/**
* @ClassName CustomConsumer
* @Description TODO
* @Author Zouhuiming
* @Date 2023/6/14 9:08
* @Version 1.0
*/publicclassCustomConsumer{publicstaticvoidmain(String[] args){//1、创建消费者的配置对象Properties properties=newProperties();//2、给消费者配置对象添加参数
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");//配置反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//配置消费者组(组内任意起名) 必须
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test2");// //修改分区分配策略// properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");//测试:修改分区分配策略ArrayList<String> startegys=newArrayList<>();
startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,startegys);//3、创建消费者对象KafkaConsumer<String,String> kafkaConsumer=newKafkaConsumer<String,String>(properties);//注册要消费的主题(可以消费多个主题)ArrayList<String> topics=newArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);//拉取数据打印while(true){//设置1s中消费一批数据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()在控制台观察生成的数据。
(3)在IDEA控制台观察收到的数据
1.3.2 独立消费者案例(订阅分区)
1、需求:创建一个独立消费者,消费first主题0号分区的数据
2、实现步骤
(1)代码编写
package org.zhm.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import javax.lang.model.type.ArrayType;import java.time.Duration;import java.util.ArrayList;import java.util.Properties;
/**
* @ClassName CustomConsumerPartition
* @Description TODO
* @Author Zouhuiming
* @Date 2023/6/14 9:18
* @Version 1.0
*/
public class CustomConsumerPartition {
public static void main(String[] args){
Properties properties=new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
//配置消费者组(必须)
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
KafkaConsumer<String,String>kafkaConsumer=new KafkaConsumer<String, String>(properties);
//消费某个主题的某个分区数据
ArrayList<TopicPartition>topicPartitions=new ArrayList<>();
topicPartitions.add(new TopicPartition("first",0));
kafkaConsumer.assign(topicPartitions);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 号分区的数据。
(3)在 IDEA 控制台,观察接收到的数据,只能消费到 0 号分区数据表示正确。
1.3.3 消费者组案例
1、需求:测试同一个主题的分区数据,只能由一个消费者组中的一个消费
2、案例实操
(1)复制一份基础消费者的代码,在 IDEA 中同时启动,即可启动同一个消费者组中的两个消费者。
(2)启动代码中的生产者发送消息,在 IDEA 控制台即可看到两个消费者在消费不同分区的数据(如果只发生到一个分区,可以在发送时增加延迟代码 Thread.sleep(2);)
(3)重新发送到一个全新的主题中,由于默认创建的主题分区数为 1,可以看到只能有一个消费者消费到数据。
1.4 生产经验——分区的分配以及再平衡
1、一个consumer group中有多个consumer组成,一个topic有多个partition组成,现在的问题是到底是由哪个consumer来消费哪个分区的数据。
2、kafka有四种主流的分区分配策略:Range、RoundRobin、Sticky、CooperativeSticky。可以通过配置参数partition.assignment.strategy,修改分区的分配策略。默认策略是Range+CooperativeSticky。kafka可以同时使用多个分区分配策略。
参数名称描述heartbeat.interval.msKafka 消费者和 coordinator 之间的心跳时间,默认 3s。该条目的值必须小于 session.timeout.ms,也不应该高于session.timeout.ms 的 1/3。session.timeout.msKafka 消费者和 coordinator 之间连接超时时间,默认 45s。超过该值,该消费者被移除,消费者组执行再平衡。max.poll.interval.ms消费者处理消息的最大时长,默认是 5 分钟。超过该值,该消费者被移除,消费者组执行再平衡。partition.assignment.strategy消 费 者 分 区 分 配 策 略 , 默 认 策 略 是 Range +CooperativeSticky。Kafka 可以同时使用多个分区分配策略。可 以 选 择 的 策 略 包 括 : Range 、 RoundRobin 、 Sticky 、CooperativeStick
1.4.1 Range以及再平衡
1、Range分区策略原理
Range是对每个topic而言的。
首先对同一个topic里面的分区按照序号进行排序,并对消费者按照字母顺序进行排序。
假如现在有 7 个分区,3 个消费者,排序后的分区将会是0,1,2,3,4,5,6;消费者排序完之后将会是C0,C1,C2。
通过partition数/consumer数来决定每个消费者应该消费几个分区,如果除不尽,那么前面几个消费者将会多消费1个分区。
注意:如果只是针对一个topic而言,c0消费者多消费1个分区影响不是很大。但是如果有N多个topic,那么针对每个topic,消费者c0都将会多消费一个分区,topic越多,c0消费者的分区会比其他消费者明显多消费n个分区。
容易产生数据倾斜
2、Range分区分配策略案例
(1)修改主题first为7个分区
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7
注意:分区数可以增加,但是不能减少
(2)复制 CustomConsumer 类,创建 CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2 组成消费者组,组名都为“test”,同时启动 3 个消费者。
(3)启动 CustomProducer 生产者,发送 500 条消息,随机发送到不同的分区。
3、Range 分区分配再平衡案例
(1)停止掉 0 号消费者,快速重新发送消息观看结果(45s 以内,越快越好)。
1 号消费者:消费到4、5、6号分区数据。
2 号消费者:消费到2、3号分区数据。
0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。
说明:0 号消费者挂掉后,消费者组需要按照超时时间 45s 来判断它是否退出,所以需要等待,时间到了 45s 后,判断它真的退出就会把任务分配给其他 broker 执行。
(2)再次重新发送消息观看结果(45s 以后)。
1 号消费者:消费到 0、1、2、3 号分区数据。
2 号消费者:消费到 4、5、6 号分区数据。
说明:消费者 0 已经被踢出消费者组,所以重新按照 range 方式分配。
1.4.2 RoundRobin以及再平衡
1、RoundRobin 分区策略原理
RoundRobin针对集群中所以Topic而言。
RoundRobin轮询分区策略,是把所有的partition和所有的consumer都列出来,然后按照hashcode进行排序,最后通过轮询算法来分配partition给到各个消费者。
1.4.3 Sticky以及再平衡
粘性分区定义:可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前,考虑上一次分配的结果,尽量少的调整分配的变动,可以节省大量的开销。
粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略,首先会尽量均衡的放置分区到消费者上面,在出现同一消费者组内消费者出现问题的时候,会尽量保持原有分配的分区不变化。
1.5 offset位移
1.5.1 offset的默认维护位置
_consumer_offsets主题里面采用key和value的方式储存数据。key是group.id+topic+分区号,value值就是当前offset的值。每隔一段时间,kafka内部会对这个topic进行compact,也就是每个group.id+topic+分区号就保留最新数据。
1.5.2 自动提交offset
为了使我们能够专注于自己的业务逻辑,Kafka提供了自动提交offset的功能。
自动提交offset的相关参数:
- enable.auto.commit:是否开启自动提交offset功能,默认是true
- auto.commit.interval.ms:自动提交offset的时间间隔,默认是5s。
1.5.3 手动提交offset
虽然自动提交offset十分简单便利,但由于其是基于时间提交的,开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
手动提交offset的方法有两种:分别是commitSync(同步提交)和commitAsync(异步提交)。两者的相同点是,都会将本次提交的一批数据最高的偏移量提交;不同点是,同步提交阻塞当前线程,一直到提交成功,并且会自动失败重试(由不可控因素导致,也会出现提交失败);而异步提交则没有失败重试机制,故有可能提交失败。
- commitSync(同步提交):必须等待offset提交完毕,再去消费下一批数据。
- commitAsync(异步提交):发送完提交offset请求后,就开始消费下一批数据。
1.5.4 指定offset消费
auto.offset.reset = earliest | latest | none 默认是 latest。
当 Kafka 中没有初始偏移量(消费者组第一次消费)或服务器上不再存在当前偏移量时(例如该数据已被删除),该怎么办?
(1)earliest:自动将偏移量重置为最早的偏移量,–from-beginning。
(2)latest(默认值):自动将偏移量重置为最新偏移量。
(3)none:如果未找到消费者组的先前偏移量,则向消费者抛出异常。
(4)任意指定 offset 位移开始消费
1.5.5 漏消费和重复消费
重复消费:已经消费了数据,但是 offset 没提交。
漏消费:先提交 offset 后消费,有可能会造成数据的漏消费。
1.6 消费者事务
如果想完成Consumer端的精准一次性消费,那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质(比 如MySQL)。
1.7 数据积压(消费者如何提高吞吐量)
1、如果是Kafka消费能力不足,则可以考虑增加Topic的分区数,并且同时提升消费组的消费者数量,消费者数 = 分区数。(两者缺一不可)
2、如果是下游的数据处理不及时:提高每批次拉取的数量。批次拉取数据过少(拉取数据/处理时间 < 生产速度),使处理的数据小于生产的数据,也会造成数据积压。
版权归原作者 星光下的赶路人star 所有, 如有侵权,请联系我们删除。