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

2024-04-12 07:04
文章标签 使用 说明 commit auto reset kafka enable offset

本文主要是介绍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。

相关文章

Zookeeper安装和配置说明

Zookeeper安装和配置说明

一、Zookeeper的搭建方式 Zookeeper安装方式有三种，单机模式和集群模式以及伪集群模式。 ■ 单机模式：Zookeeper只运行在一台服务器上，适合测试环境； ■ 伪集群模式：就是在一台物理机上运行多个Zookeeper 实例； ■ 集群模式：Zookeeper运行于一个集群上，适合生产环境，这个计算机集群被称为一个“集合体”（ensemble） Zookeeper通过复制来实现
阅读更多...
中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
使用opencv优化图片（画面变清晰）

使用opencv优化图片（画面变清晰）

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化，使其看起来更清晰，同时保持尺寸不变，通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多，主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器：相机传
阅读更多...
pdfmake生成pdf的使用

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据，将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma
阅读更多...
零基础学习Redis(10) -- zset类型命令使用

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合，内部除了存储元素外，还会存储一个score，存储在zset中的元素会按照score的大小升序排列，不同元素的score可以重复，score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]
阅读更多...
搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
git使用的说明总结

git使用的说明总结

Git使用说明 下载安装(下载地址) macOS: Git - Downloading macOS Windows: Git - Downloading Windows Linux/Unix: Git (git-scm.com) 创建新仓库 本地创建新仓库：创建新文件夹，进入文件夹目录，执行指令 git init ，用以创建新的git 克隆仓库 执行指令用以创建一个本地仓库的
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2