4.5. 在提交偏移时避免数据丢失或重复
Kafka auto-commit 机制允许 使用者自动提交消息的偏移。如果启用,消费者将从轮询代理接收的偏移以 5000ms 间隔提交。
auto-commit 机制比较方便,但它引入了数据丢失和重复的风险。如果消费者获取并转换了很多消息,但系统会在执行 auto-commit 时与消费者缓冲区中处理的消息崩溃,则该数据会丢失。如果系统在处理消息后崩溃,但在执行 auto-commit 前,数据会在重新平衡后在另一个消费者实例上重复。
auto-committing 可以避免仅在下一次轮询到代理或消费者关闭前处理所有消息时数据丢失。
要最小化数据丢失或重复的可能性,您可以将 enable.auto.commit
设置为 false
,并开发您的客户端应用程序来更好地控制提交偏移。或者,您可以使用 auto.commit.interval.ms
来减少提交之间的间隔。
... ...
# ...
enable.auto.commit=false
# ...
- 1
- 自动提交设置为 false,以提供更多对提交偏移的控制。
通过将 enable.auto.commit
设置为 false
,您可以在 执行所有 处理后提交偏移,并且消息已被使用。例如,您可以设置应用程序来调用 Kafka commitSync
和 commitAsync
提交 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
(default)
使用 read_committed
来确保仅由消费者读取提交的事务消息。但是,这会导致端到端延迟增加,因为消费者将无法返回消息,直到代理编写记录事务结果的事务标记(提交 或 中止)。
... ...
# ...
enable.auto.commit=false
isolation.level=read_committed
# ...
- 1
- 设置
read_committed
,以便只有提交的消息才会被消费者读取。