Kafka消费者

文章目录

Kafka消费者

1、消费方式

• pull（拉）模式从broker中读取数据，可以根据consumer的消费能力以适当的速率消费消息。pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。
• push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。

2、基础消费者

步骤：

1. 创建消费者的配置对象
2. 给消费者配置对象添加参数配置kafka的地址端口配置反序列化配置消费者组
3. 注册主题并让消费者订阅主题集合
4. 拉取数据打印

packagecom.hpu.kafka.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;importjava.time.Duration;importjava.util.ArrayList;importjava.util.Properties;/**
 * @author zyn
 * @version 1.0
 * @date 2021/12/29 14:30
 */publicclassMyConsumer{publicstaticvoidmain(String[] args){//1.创建消费者的配置对象Properties properties =newProperties();//2.给消费者配置对象添加参数//配置kafka的地址端口
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");//配置反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");//配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"zyn");KafkaConsumer<String,String> kafkaConsumer =newKafkaConsumer<String,String>(properties);//3.注册主题ArrayList<String> topics =newArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);//4.拉取数据打印while(true){ConsumerRecords<String,String> poll = kafkaConsumer.poll(Duration.ofSeconds(100));for(ConsumerRecord<String,String>record: poll){System.out.println(record);}}}}

3、消费者组案例

复制两份基础消费者的代码，在idea中同时启动，即可启动同一个消费者组中的三个消费者。

启动一个生产者往topic中发送数据，可以看到发往不同的分区。一个分区只会对应一个消费者。

4、分区分配策略

多个分区如何分配给各个消费者？

Kafka有三种分配策略：

• RoundRobin：轮循//消费者properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");//生产者properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"org.apache.kafka.clients.producer.RoundRobinPartitioner");
• Range(默认)：范围划分//消费者properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RangeAssignor");//生产者properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"org.apache.kafka.clients.producer.internals.DefaultPartitioner");
• Sticky：随机//消费者properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");//生产者properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"org.apache.kafka.clients.producer.UniformStickyPartitioner");

当某个消费者挂掉时，前两种需要将全部分区重新进行轮循或者范围划分出分区与对应剩余消费者的对应关系。而Sticky保持原有正常消费者的分区不变的基础上，将挂掉broker的分区随机分配给其余正常的broker。效率更高。

5、offset的维护

consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。

解决方案：

Kafka 0.9版本之前，consumer默认将offset保存在Zookeeper中，从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets。

消费offset案例

properties.put(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG,“false”);

packagecom.hpu.kafka.offset;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importjava.time.Duration;importjava.util.ArrayList;importjava.util.Properties;/**
 * @author zyn
 * @version 1.0
 * @date 2021/12/29 20:25
 */publicclassMyOffset{publicstaticvoidmain(String[] args){Properties properties =newProperties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"offset");

        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG,"false");KafkaConsumer<Object,Object> kafkaConsumer =newKafkaConsumer<>(properties);ArrayList<String> topics =newArrayList<>();
        topics.add("__consumer_offsets");
        kafkaConsumer.subscribe(topics);while(true){ConsumerRecords<Object,Object> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<Object,Object> consumerRecord : consumerRecords){System.out.println(consumerRecord);}}}}

#指定消费者组
kafka-consumer-groups.sh  --bootstrap-server hadoop102:9092 --group testTopic  --describe
#全部消费者组和全部主题
kafka-consumer-groups.sh --all-groups --all-topics --describe --bootstrap-server hadoop102:9092

6、自动提交offset

KafkaConsumer：需要创建一个消费者对象，用来消费数据

ConsumerConfig：获取所需的一系列配置参数

ConsuemrRecord：每条数据都要封装成一个ConsumerRecord对象

为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。

自动提交offset的相关参数：

enable.auto.commit：是否开启自动提交offset功能

auto.commit.interval.ms：自动提交offset的时间间隔

importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importjava.util.Arrays;importjava.util.Properties;/**
 * 1. 创建消费者配置类
 * 2. 添加配置
 * 3. 创建消费者对象
 * 4. 设置消费的主题
 * 5. 挂起消费数据
 */publicclassCustomConsumer{publicstaticvoidmain(String[] args){// 1. 创建kafka消费者配置类Properties properties =newProperties();// 2. 添加配置参数// 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");// 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");// 配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 是否自动提交offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");// 提交offset的时间周期
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//3. 创建kafka消费者KafkaConsumer<String,String> consumer =newKafkaConsumer<>(properties);//4. 设置消费主题  形参是列表
        consumer.subscribe(Arrays.asList("first"));//5. 消费数据while(true){// 读取消息ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));// 输出消息for(ConsumerRecord<String,String> consumerRecord : consumerRecords){System.out.println(consumerRecord.value());}}}}

7、重置offset

auto.offset.reset = earliest | latest | none |

（1）earliest：自动将偏移量重置为最早的偏移量

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

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

8、手动提交offset

由于其是基于时间提交的，开发人员难以把握offset提交的时机。手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次poll的一批数据最高的偏移量提交；不同点是，commitSync阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败。

同步提交

将自动提交关闭：

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, “false”);

开启同步提交：

consumer.commitSync();

importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importjava.util.Arrays;importjava.util.Properties;/**
 * 1. 修改自动提交offset为手动
 * 2. 在业务代码完成之后手动同步提交offset
 */publicclassCustomConsumerByHand{publicstaticvoidmain(String[] args){// 1. 创建kafka消费者配置类Properties properties =newProperties();// 2. 添加配置参数// 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");// 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");// 配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 是否自动提交offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");// 提交offset的时间周期
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//3. 创建kafka消费者KafkaConsumer<String,String> consumer =newKafkaConsumer<>(properties);//4. 设置消费主题  形参是列表
        consumer.subscribe(Arrays.asList("first"));//5. 消费数据while(true){// 读取消息ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));// 输出消息for(ConsumerRecord<String,String> consumerRecord : consumerRecords){System.out.println(consumerRecord.value());}// 同步提交offset
            consumer.commitSync();}}}

异步提交

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

importorg.apache.kafka.clients.consumer.*;importorg.apache.kafka.common.TopicPartition;importjava.util.Arrays;importjava.util.Map;importjava.util.Properties;/**
 * 1. 修改自动提交offset为手动
 * 2. 在业务代码完成之后手动异步提交offset
 */publicclassCustomConsumerByHand{publicstaticvoidmain(String[] args){// 1. 创建kafka消费者配置类Properties properties =newProperties();// 2. 添加配置参数// 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");// 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");// 配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 是否自动提交offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");// 提交offset的时间周期
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//3. 创建kafka消费者KafkaConsumer<String,String> consumer =newKafkaConsumer<>(properties);//4. 设置消费主题  形参是列表
        consumer.subscribe(Arrays.asList("first"));//5. 消费数据while(true){// 读取消息ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));// 输出消息for(ConsumerRecord<String,String> consumerRecord : consumerRecords){System.out.println(consumerRecord.value());}// 同步提交offset//consumer.commitSync();// 异步提交offset
            consumer.commitAsync(newOffsetCommitCallback(){/**
                 * 回调函数输出
                 * @param offsets   offset信息
                 * @param exception 异常
                 */@OverridepublicvoidonComplete(Map<TopicPartition,OffsetAndMetadata> offsets,Exception exception){// 如果出现异常打印if(exception !=null){System.err.println("Commit failed for "+ offsets);}else{Set<TopicPartition> topicPartitions = offsets.keySet();//该方法可以得到我们消费的的消息 所处的topic partition 有哪些for(TopicPartition topicPartition : topicPartitions){//遍历我们消费的topic 以及parition 元数据OffsetAndMetadata offsetAndMetadata = offsets.get(topicPartition);//每个topic 的每个parition 的消费到的offsetlong offset = offsetAndMetadata.offset();//获取提交的offset值int partition = topicPartition.partition();//获取该parttion的值String topic = topicPartition.topic();//获取该toipic的值System.out.printf("----提交的offset = %s, 该 partition = %s ,以及topic = %s---------\n",
                                    offset,partition,topic);}}}});}}}

9、Consumer事务

无论是同步提交还是异步提交offset，都有可能会造成数据的漏消费或者重复消费。先提交offset后消费，有可能造成数据的漏消费；而先消费后提交offset，有可能会造成数据的重复消费。

如果想完成Consumer端的精准一次性消费，那么需要kafka消费端将消费过程和提交offset过程做原子绑定，要么都成功，要么都失败。此时我们需要将kafka的offset保存到支持事务的自定义介质（比如mysql）。