kafka消费者消息确认

Apache Kafka是一个分布式流处理平台，被广泛用于建立实时数据流应用程序和数据管道。在Kafka中，有消费者这样的角色用于接收并处理存储在Kafka中的消息。本文将介绍如何在Kafka消费者中确认消息，以及如何定期提交偏移量，以确保在消费者和broker之间正确同步消息。

1. 确认消费

Kafka提供了两种方式来确认消费者是否已经消费了消息：

• 自动确认模式（AUTO_COMMIT_INTERVAL_MS_CONFIG）：消费者配置中的auto.commit.interval.ms属性将控制确认消息的自动提交。在这种情况下，每隔一段时间（通常是一秒钟），消费者将自动确认已消费的消息。
• 手动确认模式（enable.auto.commit为false）：在这种确认模式下，消费者需要在处理完消息之后显式地调用确认API来通知broker消息已经被消费了，以便broker可以删除消息。确认API有三种形式：

// 确认指定的偏移量
consumer.commitSync(Collections.singletonMap(
    new TopicPartition("topic", 0), new OffsetAndMetadata(100)));

// 确认偏移量直到最后一条未确认的消息（异步确认）
consumer.commitAsync();

// 确认当前消费者的进度（异步确认），包括未确认的消息
consumer.commitAsync(currentOffsets, null);

请注意，在手动确认模式下需要小心，否则可能会出现消费者无法获取消息的情况。处理方法是在必要时记录已经处理的消息并在消费者重新启动时恢复。

2. 提交偏移量

消费者需要定期提交偏移量，以确保在消费者和broker之间正确同步消息。如果偏移量没有提交，则消费者可能会重复消费一些消息，或者在已确认的消息应该被删除时将其留在broker上。

在自动确认模式下，偏移量将在一段时间后自动提交，以便消费者可以重试在失败时重试操作。在手动确认模式下，消费者需要显式地提交偏移量。这可以通过以下方式完成：

consumer.commitAsync();

或者

consumer.commit