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。
总结：

1. package com.tencent.tcb.operation.ckafka.plain;
4. import com.google.common.collect.Lists;
5. import com.tencent.tcb.operation.ckafka.JavaKafkaConfigurer;
6. import java.time.Instant;
7. import java.time.temporal.ChronoUnit;
8. import java.util.ArrayList;
9. import java.util.Collection;
10. import java.util.HashMap;
11. import java.util.List;
12. import java.util.Map;
13. import java.util.Map.Entry;
14. import java.util.Properties;
15. import org.apache.kafka.clients.CommonClientConfigs;
16. import org.apache.kafka.clients.consumer.ConsumerConfig;
17. import org.apache.kafka.clients.consumer.ConsumerRecord;
18. import org.apache.kafka.clients.consumer.ConsumerRecords;
19. import org.apache.kafka.clients.consumer.KafkaConsumer;
20. import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
21. import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
22. import org.apache.kafka.clients.producer.ProducerConfig;
23. import org.apache.kafka.common.PartitionInfo;
24. import org.apache.kafka.common.TopicPartition;
25. import org.apache.kafka.common.config.SaslConfigs;
28. public class KafkaPlainConsumerDemo {
31.     public static void main(String args[]) {
32.         //设置JAAS配置文件的路径。
33.         JavaKafkaConfigurer.configureSaslPlain();
36.         //加载kafka.properties。
37.         Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
40.         Properties props = new Properties();
41.         //设置接入点，请通过控制台获取对应Topic的接入点。
42.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
45.         //接入协议。
46.         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
47.         //Plain方式。
48.         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
49.         //两次Poll之间的最大允许间隔。
50.         //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
51.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
52.         //每次Poll的最大数量。
53.         //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
54.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
55.         //消息的反序列化方式。
56.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
57.                 "org.apache.kafka.common.serialization.StringDeserializer");
58.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
59.                 "org.apache.kafka.common.serialization.StringDeserializer");
60.         //当前消费实例所属的消费组，请在控制台申请之后填写。
61.         //属于同一个组的消费实例，会负载消费消息。
62.         props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
65.         //消费offset的位置。注意！如果auto.offset.reset=none这样设置，消费组在第一次消费的时候 就会报错找不到offset，第一次这时候就需要在catch里手动设置offset。
66.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
67.         //构造消费对象，也即生成一个消费实例。
68.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
69.         //设置消费组订阅的Topic，可以订阅多个。
70.         //如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
71.         List<String> subscribedTopics = new ArrayList<String>();
72.         //如果需要订阅多个Topic，则在这里添加进去即可。
73.         //每个Topic需要先在控制台进行创建。
74.         String topicStr = kafkaProperties.getProperty("topic");
75.         String[] topics = topicStr.split(",");
76.         for (String topic : topics) {
77.             subscribedTopics.add(topic.trim());
78.         }
79.         consumer.subscribe(subscribedTopics);
80.         //循环消费消息。
81.         while (true) {
82.             try {
83.                 ConsumerRecords<String, String> records = consumer.poll(1000);
84.                 //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。 建议开一个单独的线程池来消费消息，然后异步返回结果。
85.                 for (ConsumerRecord<String, String> record : records) {
86.                     System.out.println(
87.                             String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
88.                 }
89.             } catch (NoOffsetForPartitionException e) {
90.                 System.out.println(e.getMessage());
93.                 //当auto.offset.reset设置为 none时，需要捕获异常 自己设置offset。您可以根据自身业务情况选择以下方式中的其中一种。
94.                 //e.g 1 :指定offset, 这里需要自己维护offset，方便重试。
95.                 Map<Integer, Long> partitionBeginOffsetMap = getPartitionOffset(consumer, topicStr, true);
96.                 Map<Integer, Long> partitionEndOffsetMap = getPartitionOffset(consumer, topicStr, false);
97.                 consumer.seek(new TopicPartition(topicStr, 0), 0);
100.                 //e.g 2:从头开始消费
101.                 consumer.seekToBeginning(Lists.newArrayList(new TopicPartition(topicStr, 0)));
104.                 //e.g 3:指定offset为最近可用的offset。
105.                 consumer.seekToEnd(Lists.newArrayList(new TopicPartition(topicStr, 0)));
108.                 //e.g 4: 根据时间戳获取offset，就是根据时间戳去设置offset。例如重置到10分钟前的offset
109.                 Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
110.                 Long value = Instant.now().minus(300, ChronoUnit.SECONDS).toEpochMilli();
111.                 timestampsToSearch.put(new TopicPartition(topicStr, 0), value);
112.                 Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = consumer
113.                         .offsetsForTimes(timestampsToSearch);
114.                 for (Entry<TopicPartition, OffsetAndTimestamp> entry : topicPartitionOffsetAndTimestampMap
115.                         .entrySet()) {
116.                     TopicPartition topicPartition = entry.getKey();
117.                     OffsetAndTimestamp entryValue = entry.getValue();
118.                     consumer.seek(topicPartition, entryValue.offset()); // 指定offset, 这里需要自己维护offset，方便重试。
119.                 }
124.             }
125.         }
126.     }
129.     /**
130.      * 获取topic的最早、最近的offset
131.      * @param consumer
132.      * @param topicStr
133.      * @param beginOrEnd true begin; false end
134.      * @return
135.      */
136.     private static Map<Integer, Long> getPartitionOffset(KafkaConsumer<String, String> consumer, String topicStr,
137.             boolean beginOrEnd) {
138.         Collection<PartitionInfo> partitionInfos = consumer.partitionsFor(topicStr);
139.         List<TopicPartition> tp = new ArrayList<>();
140.         Map<Integer, Long> map = new HashMap<>();
141.         partitionInfos.forEach(str -> tp.add(new TopicPartition(topicStr, str.partition())));
142.         Map<TopicPartition, Long> topicPartitionLongMap;
143.         if (beginOrEnd) {
144.             topicPartitionLongMap = consumer.beginningOffsets(tp);
145.         } else {
146.             topicPartitionLongMap = consumer.endOffsets(tp);
147.         }
148.         topicPartitionLongMap.forEach((key, beginOffset) -> {
149.             int partition = key.partition();
150.             map.put(partition, beginOffset);
151.         });
152.         return map;
153.     }
156. }

复制代码

springboot项目下

1. /**
2.      * enable-auto-commit: false 由spring提交
3.      * enable-auto-commit: true  由kafka提交
4.      */
6.     /**
7.      * enable-auto-commit: true  相同组下  （换组 会重置数据）
8.      * 如果这个topic某个分区有已经提交的offset,那么无论是把auto.offset.reset=earliest还是latest,都将失效,消费者会从已经提交的offset开始消费.
9.      */
11.     /**
12.      * enable-auto-commit: fasle 相同组下 （换组 会重置数据）
13.      * 如果这个topic某个分区没有提交的offset,那么把auto.offset.reset=latest,将没消费的设置为提交消费，然后从最后开始消费
14.      * 如果这个topic某个分区没有提交的offset,那么把auto.offset.reset=earliest,从没开始消费的offset开始消费
15.      */

复制代码

非springboot项目下

1. enable.auto.commit false
2.     auto.offset.reset earliest 第一次消费， 重启后消费  都会从第一条开始重新消费全部数据
3. enable.auto.commit true
4.     auto.offset.reset earliest 第一次消费全部数据，重启后从提交处开始消费
6. enable.auto.commit false
7.     auto.offset.reset latest  第一次，重启后会从最后一条开始消费，但没有提交，换成earliest 重新消费全部数据
8. enable.auto.commit true
9.    auto.offset.reset latest   第一次从最后一条开始消费,重启后从提交处开始消费

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。