4.5. 在提交偏移时避免数据丢失或重复

Kafka auto-commit 机制 允许使用者自动提交消息偏移。如果启用,消费者将提交从以 5000ms 间隔轮询代理接收的偏移量。

auto-commit 机制很方便,但会带来数据丢失和重复的风险。如果消费者获取并转换了多个消息,但系统在执行自动提交时会在消费者缓冲区中崩溃处理的消息,则该数据将会丢失。如果系统在处理消息后崩溃,但在执行 auto-commit 之前,数据会在重新平衡后在另一个消费者实例上重复。

只有在下一次轮询代理或消费者关闭前处理所有信息时,自动提交才能避免数据丢失。

要最小化数据丢失或重复的可能性,您可以将 enable.auto.commit 设置为 false,并开发客户端应用来更好地控制提交偏移。或者,您可以使用 auto.commit.interval.ms 来降低提交之间的间隔。

# ...
enable.auto.commit=false 1
# ...
1
自动提交设置为 false,以提供对提交偏移的更多控制。

通过将 enable.auto.commit 设置为 false,您可以在 执行所有 处理后提交偏移,并且消息已被使用。例如,您可以设置应用程序来调用 Kafka commitSynccommitAsync 提交 API。

commitSync API 在从轮询返回的消息批处理中提交偏移量。完成处理批处理后,您将调用 API。如果使用 commitSync API,则应用程序不会轮询新消息,直到批处理中的最后一个偏移提交为止。如果这个负面影响吞吐量,您可以更频繁提交,也可以使用 commitAsync API。commitAsync API 不会等待代理响应提交请求,但风险在重新平衡时创建更多重复。一个常见的方法是,将应用程序中的提交 API 与仅在关闭消费者或重新平衡前使用的 commitSync API 组合,以确保最终提交成功。

4.5.1. 控制事务消息

考虑在制作者一侧使用事务 ID 和启用 idempotence (enable.idempotence=true)来保证完全传输。然后,您可以使用 isolation.level 属性来控制使用者读取事务消息的方式。

isolation.level 属性有两个有效值:

  • read_committed
  • read_uncommitted (默认)

使用 read_committed 来确保只有提交的事务消息才会被消费者读取。但是,这将增加端到端延迟,因为消费者将无法返回消息,直到代理编写了记录事务结果的事务标记(提交中止)。

# ...
enable.auto.commit=false
isolation.level=read_committed 1
# ...
1
设置 read_committed,以便只有提交的消息才会被消费者读取。