kafka enable.auto.commit和auto.offset.reset使用说明

2024-04-12 07:04

本文主要是介绍kafka enable.auto.commit和auto.offset.reset使用说明,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

enable.auto.commit

是否自动提交offset,默认是true。

auto.offset.reset

表示自动重置 offset。

auto.offset.reset 参数定义了当无法获取消费分区的位移时从何处开始消费。例如:当 Broker 端没有 offset(如第一次消费或 offset 超过7天过期)时如何初始化 offset,当收到 OFFSET_OUT_OF_RANGE 错误时如何重置 Offset。

earliest:自动重置到 partition 的最小 offset。
当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费。
latest:默认为 latest,表示自动重置到 partition 的最大 offset。
当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据。
none:不自动进行 offset 重置,抛出 OffsetOutOfRangeException 异常。
topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常。

auto.offset.reset=none 使用说明

使用背景

不希望发生 offset 自动重置的情况,因为业务不允许发生大规模的重复消费。

注意:

此时消费组在第一次消费的时候就会找不到 offset 而报错,这时就需要在 catch 里手动设置 offset。

使用说明

auto.offset.reset 设置为 None 以后,可以避免 offset 自动重置的问题,但是当增加分区的时候,因为关闭了自动重置机制,客户端不知道新的分区要从哪里开始消费,则会产生异常,此时需要人工去设置消费分组 offset 并消费。

使用方式

消费者在消费时,当 consumer 设置 auto.offset.reset=none, 捕获到 NoOffsetForPartitionException 异常,在 catch 里自己设置 offset。您可以根据自身业务情况选择以下方式中的其中一种。

指定 offset,这里需要自己维护 offset,方便重试。

指定从头开始消费。

指定 offset 为最近可用的 offset。

根据时间戳获取 offset,设置 offset。

总结:

package com.tencent.tcb.operation.ckafka.plain;import com.google.common.collect.Lists;
import com.tencent.tcb.operation.ckafka.JavaKafkaConfigurer;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;
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.consumer.NoOffsetForPartitionException;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.SaslConfigs;public class KafkaPlainConsumerDemo {public static void main(String args[]) {//设置JAAS配置文件的路径。JavaKafkaConfigurer.configureSaslPlain();//加载kafka.properties。Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();Properties props = new Properties();//设置接入点,请通过控制台获取对应Topic的接入点。props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));//接入协议。props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");//Plain方式。props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");//两次Poll之间的最大允许间隔。//消费者超过该值没有返回心跳,服务端判断消费者处于非存活状态,服务端将消费者从Consumer Group移除并触发Rebalance,默认30s。props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);//每次Poll的最大数量。//注意该值不要改得太大,如果Poll太多数据,而不能在下次Poll之前消费完,则会触发一次负载均衡,产生卡顿。props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);//消息的反序列化方式。props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");//当前消费实例所属的消费组,请在控制台申请之后填写。//属于同一个组的消费实例,会负载消费消息。props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));//消费offset的位置。注意!如果auto.offset.reset=none这样设置,消费组在第一次消费的时候 就会报错找不到offset,第一次这时候就需要在catch里手动设置offset。props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");//构造消费对象,也即生成一个消费实例。KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);//设置消费组订阅的Topic,可以订阅多个。//如果GROUP_ID_CONFIG是一样,则订阅的Topic也建议设置成一样。List<String> subscribedTopics = new ArrayList<String>();//如果需要订阅多个Topic,则在这里添加进去即可。//每个Topic需要先在控制台进行创建。String topicStr = kafkaProperties.getProperty("topic");String[] topics = topicStr.split(",");for (String topic : topics) {subscribedTopics.add(topic.trim());}consumer.subscribe(subscribedTopics);//循环消费消息。while (true) {try {ConsumerRecords<String, String> records = consumer.poll(1000);//必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。 建议开一个单独的线程池来消费消息,然后异步返回结果。for (ConsumerRecord<String, String> record : records) {System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));}} catch (NoOffsetForPartitionException e) {System.out.println(e.getMessage());//当auto.offset.reset设置为 none时,需要捕获异常 自己设置offset。您可以根据自身业务情况选择以下方式中的其中一种。//e.g 1 :指定offset, 这里需要自己维护offset,方便重试。Map<Integer, Long> partitionBeginOffsetMap = getPartitionOffset(consumer, topicStr, true);Map<Integer, Long> partitionEndOffsetMap = getPartitionOffset(consumer, topicStr, false);consumer.seek(new TopicPartition(topicStr, 0), 0);//e.g 2:从头开始消费consumer.seekToBeginning(Lists.newArrayList(new TopicPartition(topicStr, 0)));//e.g 3:指定offset为最近可用的offset。consumer.seekToEnd(Lists.newArrayList(new TopicPartition(topicStr, 0)));//e.g 4: 根据时间戳获取offset,就是根据时间戳去设置offset。例如重置到10分钟前的offsetMap<TopicPartition, Long> timestampsToSearch = new HashMap<>();Long value = Instant.now().minus(300, ChronoUnit.SECONDS).toEpochMilli();timestampsToSearch.put(new TopicPartition(topicStr, 0), value);Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = consumer.offsetsForTimes(timestampsToSearch);for (Entry<TopicPartition, OffsetAndTimestamp> entry : topicPartitionOffsetAndTimestampMap.entrySet()) {TopicPartition topicPartition = entry.getKey();OffsetAndTimestamp entryValue = entry.getValue();consumer.seek(topicPartition, entryValue.offset()); // 指定offset, 这里需要自己维护offset,方便重试。}}}}/*** 获取topic的最早、最近的offset* @param consumer* @param topicStr* @param beginOrEnd true begin; false end* @return*/private static Map<Integer, Long> getPartitionOffset(KafkaConsumer<String, String> consumer, String topicStr,boolean beginOrEnd) {Collection<PartitionInfo> partitionInfos = consumer.partitionsFor(topicStr);List<TopicPartition> tp = new ArrayList<>();Map<Integer, Long> map = new HashMap<>();partitionInfos.forEach(str -> tp.add(new TopicPartition(topicStr, str.partition())));Map<TopicPartition, Long> topicPartitionLongMap;if (beginOrEnd) {topicPartitionLongMap = consumer.beginningOffsets(tp);} else {topicPartitionLongMap = consumer.endOffsets(tp);}topicPartitionLongMap.forEach((key, beginOffset) -> {int partition = key.partition();map.put(partition, beginOffset);});return map;}}

springboot项目下

 /*** enable-auto-commit: false 由spring提交* enable-auto-commit: true  由kafka提交*//*** enable-auto-commit: true  相同组下  (换组 会重置数据)* 如果这个topic某个分区有已经提交的offset,那么无论是把auto.offset.reset=earliest还是latest,都将失效,消费者会从已经提交的offset开始消费.*//*** enable-auto-commit: fasle 相同组下 (换组 会重置数据)* 如果这个topic某个分区没有提交的offset,那么把auto.offset.reset=latest,将没消费的设置为提交消费,然后从最后开始消费* 如果这个topic某个分区没有提交的offset,那么把auto.offset.reset=earliest,从没开始消费的offset开始消费*/

非springboot项目下

enable.auto.commit falseauto.offset.reset earliest 第一次消费, 重启后消费  都会从第一条开始重新消费全部数据
enable.auto.commit trueauto.offset.reset earliest 第一次消费全部数据,重启后从提交处开始消费enable.auto.commit falseauto.offset.reset latest  第一次,重启后会从最后一条开始消费,但没有提交,换成earliest 重新消费全部数据
enable.auto.commit trueauto.offset.reset latest   第一次从最后一条开始消费,重启后从提交处开始消费

这篇关于kafka enable.auto.commit和auto.offset.reset使用说明的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/896427

相关文章

Linux换行符的使用方法详解

《Linux换行符的使用方法详解》本文介绍了Linux中常用的换行符LF及其在文件中的表示,展示了如何使用sed命令替换换行符,并列举了与换行符处理相关的Linux命令,通过代码讲解的非常详细,需要的... 目录简介检测文件中的换行符使用 cat -A 查看换行符使用 od -c 检查字符换行符格式转换将

使用Jackson进行JSON生成与解析的新手指南

《使用Jackson进行JSON生成与解析的新手指南》这篇文章主要为大家详细介绍了如何使用Jackson进行JSON生成与解析处理,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1. 核心依赖2. 基础用法2.1 对象转 jsON(序列化)2.2 JSON 转对象(反序列化)3.

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》:本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.

Elasticsearch 在 Java 中的使用教程

《Elasticsearch在Java中的使用教程》Elasticsearch是一个分布式搜索和分析引擎,基于ApacheLucene构建,能够实现实时数据的存储、搜索、和分析,它广泛应用于全文... 目录1. Elasticsearch 简介2. 环境准备2.1 安装 Elasticsearch2.2 J

使用C#代码在PDF文档中添加、删除和替换图片

《使用C#代码在PDF文档中添加、删除和替换图片》在当今数字化文档处理场景中,动态操作PDF文档中的图像已成为企业级应用开发的核心需求之一,本文将介绍如何在.NET平台使用C#代码在PDF文档中添加、... 目录引言用C#添加图片到PDF文档用C#删除PDF文档中的图片用C#替换PDF文档中的图片引言在当

Java中List的contains()方法的使用小结

《Java中List的contains()方法的使用小结》List的contains()方法用于检查列表中是否包含指定的元素,借助equals()方法进行判断,下面就来介绍Java中List的c... 目录详细展开1. 方法签名2. 工作原理3. 使用示例4. 注意事项总结结论:List 的 contain

C#使用SQLite进行大数据量高效处理的代码示例

《C#使用SQLite进行大数据量高效处理的代码示例》在软件开发中,高效处理大数据量是一个常见且具有挑战性的任务,SQLite因其零配置、嵌入式、跨平台的特性,成为许多开发者的首选数据库,本文将深入探... 目录前言准备工作数据实体核心技术批量插入:从乌龟到猎豹的蜕变分页查询:加载百万数据异步处理:拒绝界面

Android中Dialog的使用详解

《Android中Dialog的使用详解》Dialog(对话框)是Android中常用的UI组件,用于临时显示重要信息或获取用户输入,本文给大家介绍Android中Dialog的使用,感兴趣的朋友一起... 目录android中Dialog的使用详解1. 基本Dialog类型1.1 AlertDialog(

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意

使用Sentinel自定义返回和实现区分来源方式

《使用Sentinel自定义返回和实现区分来源方式》:本文主要介绍使用Sentinel自定义返回和实现区分来源方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Sentinel自定义返回和实现区分来源1. 自定义错误返回2. 实现区分来源总结Sentinel自定