12.4.2.5. 在提交偏移时避免数据丢失或重复
Kafka 自动提交机制 允许使用者自动提交消息偏移。如果启用,消费者将以 5000ms 间隔提交从轮询代理收到的偏移。
自动提交机制很方便,但会带来数据丢失和复制风险。如果使用者已获取并转换了大量消息,但执行自动提交时,系统会对消费者缓冲区中已处理的消息崩溃,该数据将会丢失。如果系统在处理消息后崩溃,但在执行自动使用前,数据会在重新平衡后在另一个消费者实例上重复。
仅当在下一次轮询到代理或消费者关闭之前处理所有消息时,自动使用才可以避免数据丢失。
为最大程度减少数据丢失或重复的可能性,您可以将 enable.auto.commit
设置为 false
,并开发客户端应用程序,使其对提交的偏移拥有更多控制权。或者,您可以使用 auto.commit.interval.ms
减少提交之间的间隔。
# ...
enable.auto.commit=false 1
# ...
- 1
- 自动提交设为 false,以提供对提交偏移的更多控制。
通过将 enable.auto.commit
设置为 false
,您可以在执行了 所有 处理并且消息已被使用后提交偏移。例如,您可以将应用程序设置为调用 Kafka commitSync 和 commit
Async
提交 API。
commitSync
API 在从轮询返回的消息批处理中提交偏移。完成批处理中的所有消息后,您将调用 API。如果使用 commitSync
API,则应用不会轮询新消息,直到提交批处理中的最后一个偏移。如果这会对吞吐量造成负面影响,您可以降低提交的频率,也可以使用 commitAsync
API。commitAsync
API 不等待代理响应提交请求,但可能会在重新平衡时造成更多的重复。种常见的做法是将应用中的两个提交 API 与刚才在关闭使用者或重新平衡之前使用的 commitSync
API 相结合,以确保最终提交成功。
12.4.2.5.1. 控制事务性消息
考虑在制作者一侧使用事务 id 和启用幂等性(enable.idempotence=true
)来保证准确交付。在消费者方面,您可以使用 isolated .level
属性来控制消费者如何读取事务性消息。
isolated .level
属性有两个有效值:
-
read_committed
-
read_uncomcommit
(默认)
使用 read_comsigned 来确保
只有提交的事务消息会被使用者读取。但是,这会导致端到端延迟增加,因为在代理写入了记录事务结果的事务标记(已承诺 或中止)之前,消费者将无法返回消息 。
# ...
enable.auto.commit=false
isolation.level=read_committed 1
# ...
- 1
- 设置为
read_com
commit,以使使用者只读取提交的邮件。