Kafka如何在消息处理成功之前不移动偏移量？

可以通过手动提交偏移量来实现在消息处理成功之前不移动偏移量的需求。代码示例如下：

public class KafkaConsumerExample { public static void main(String[] args) {

  String topicName = "test";
  String groupName = "group1";
  Properties props = new Properties();

  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", groupName);
  props.put("enable.auto.commit", "false");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  consumer.subscribe(Arrays.asList(topicName));

  try {
      while (true) {
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
          for (ConsumerRecord<String, String> record : records) {
              System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
              
              // process the message here

              // manually commit the offset after the message is processed successfully               
              consumer.commitSync(Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)));
          }
      }
  } finally {
      consumer.close();
  }

} }

在上面的代码中，我们将 enable.auto.commit 设置为 false，这样消费者就不会自动提交偏移量。在处理每条消息之后，我们手动提交偏移量，设置偏移量为当前消息的 offset 加 1，这样下次拉取数据时就能从上次成功处理的消息后开始读取。如果消息处理失败，则不会提交偏移量，这样消费者就会重复消费该消息，直到处理成功为止。