Kafka中的延迟包消费问题

解决延迟包消费的方法有两种，分别是手动提交偏移量和使用自动提交偏移量。

手动提交偏移量：

while(true) {
    ConsumerRecords<String, String>> records = consumer.poll(Duration.of(100, TimeUnit.MILLISECONDS));
    for (ConsumerRecord<String, String> record: records) {
        // 处理消息
    }
    consumer.commitSync(); // 手动提交偏移量
}

使用自动提交偏移量：

while(true) {
  ConsumerRecords<String, String>> records = consumer.poll(Duration.of(100, TimeUnit.MILLISECONDS));
  for (ConsumerRecord<String, String> record: records) {
      // 处理消息
  }
}
// 在退出消费者之前自动提交偏移量
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    consumer.close();
}));

在使用自动提交偏移量时，需要注意：

• 如果Kafka消费者在处理消息时出现故障，已接收但尚未处理的所有消息都会重新处理。
• 如果发生强制停止消费者，而没有处理完所有未提交的消息，那么所有未提交的消息都将被处理。

因此，手动提交偏移量更加灵活和可靠，适用于对消息处理顺序和一致性要求更高的场景。