我正在阅读这个:
自动提交最简单的提交偏移量的方法是允许消费者代替你进行提交。如果配置了enable.auto.commit=true，那么每五秒钟，消费者将提交客户端从poll()接收到的最大偏移量。五秒间隔是默认值，并通过设置auto.commit.interval.ms来控制。就像消费者中的其他所有内容一样，自动提交由轮询循环驱动。每次轮询时，消费者会检查是否到了提交的时间，如果是，它将提交上次轮询返回的偏移量。
也许我的英语不好，但我没有完全理解这个描述。
假设我使用默认间隔-5秒自动提交，并且每7秒进行一次轮询。在这种情况下，提交将每5秒或每7秒发生一次？
如果每3秒进行一次轮询，您能澄清行为吗？提交每5秒还是每6秒发生一次？
我已经阅读了这个：
而且这与答案相矛盾。
你能详细解释一下吗？
假设我有这样的图表：
0秒 - 轮询
4秒 - 轮询
8秒 - 轮询
偏移量将在何时提交，在何时提交哪个偏移量？
自动提交检查会在每次轮询时调用，并检查经过的时间是否大于配置的时间。如果是，就会提交偏移量。
如果提交间隔为5秒，而轮询发生在7秒后，则提交仅会在7秒后发生。
在poll中有一个地方调用了存储，具体位置为https://github.com/apache/kafka/blob/10cd98cc894b88c5d1e24fc54c66361ad9914df2/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L279
话虽如此，例如每7秒执行一次poll，并且将自动提交设置为5：
0 - poll，+将截止时间设置为第5秒
7 - poll + 提交截止时间，将截止时间更新为7+5=12
14 - poll + 提交截止时间，将截止时间更新为12+5=17
但是，如果将轮询设置为每3秒，并且将自动提交设置为5：
0 - poll，+将截止时间设置为第5秒
3 - poll，没有提交
6 - poll + 提交截止时间，将截止时间更新为6+5=11
public class KafkaTest {
public static final String KAFKA_TOPIC_NAME = "kafka-xx-test-topic";
public static final String CONSUMER_GROUP_ID = "test-consumer-xx";
public static final String BOOTSTRAP_SERVERS = "localhost:9092";
public static void main(String[] args) {
final KafkaProducer<Object, Object> kafkaProducer = new KafkaProducer<>(getProps());
for (int i = 0; i < 1000; i++) {
kafkaProducer.send(new ProducerRecord<>(KAFKA_TOPIC_NAME, "Data_" + i));
}
final Consumer<Long, String> consumer = new KafkaConsumer<>(getProps());
consumer.subscribe(Collections.singletonList(KAFKA_TOPIC_NAME));
TopicPartition actualTopicPartition = new TopicPartition(KAFKA_TOPIC_NAME, 0);
while (true) {
final ConsumerRecords<Long, String> consumerRecords = consumer.poll(Duration.ofSeconds(60));
consumerRecords.forEach(record -> {
try {
TimeUnit.MILLISECONDS.sleep(200);
} catch (InterruptedException e) {
}
});
final long committedOffset = consumer.committed(Collections.singleton(actualTopicPartition)).get(actualTopicPartition).offset();
final long consumerCurrentOffset = consumer.position(actualTopicPartition);
System.out.println("Poll finish.. consumer-offset: " + consumerCurrentOffset + " - committed-offset: " + committedOffset + " " + LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")));
}
}
private static Map<String, Object> getProps() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // Default: latest
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // Default: true
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10); // Default: 500
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000); // Default: 5000
return props;
}
}
Poll finish.. consumer-offset: 1010 - committed-offset: 1000 17:07:05
Poll finish.. consumer-offset: 1020 - committed-offset: 1000 17:07:07
Poll finish.. consumer-offset: 1030 - committed-offset: 1000 17:07:09
Poll finish.. consumer-offset: 1040 - committed-offset: 1030 17:07:11 -> commit when poll finish because of elapsed time(6 sec) > commit interval(5 sec)
Poll finish.. consumer-offset: 1050 - committed-offset: 1030 17:07:13
Poll finish.. consumer-offset: 1060 - committed-offset: 1030 17:07:15
Poll finish.. consumer-offset: 1070 - committed-offset: 1060 17:07:17 -> auto commit
Poll finish.. consumer-offset: 1080 - committed-offset: 1060 17:07:19
Poll finish.. consumer-offset: 1090 - committed-offset: 1060 17:07:21
Poll finish.. consumer-offset: 1100 - committed-offset: 1090 17:07:23 -> auto commit
auto-commit-interval: 20000
auto-offset-reset: latest
max-poll-records: 10
max-poll-interval-ms: 20000
auto-commit-interval: 2000
auto-offset-reset: latest
max-poll-records: 10
max-poll-interval-ms: 20000