kafka位移提交问题

kafka位移提交问题,第1张

kafka位移提交问题

自动提交
(系统默认值为5秒,缩短位移提交间隔,可以更好防止数据丢失和重复消费问题)
// 自动提交,默认true

props.put("enable.auto.commit", "true");
// 设置自动每1s提交一次
props.put("auto.commit.interval.ms", "1000");

自动提交是在轮询中进行,消费者每次轮询时都会检查是否提交该偏移量。这种情况就会发生重复消费和丢失消息的情况。
手动提交(consumer.commitSync())

// 关闭自动提交
props.put("enable.auto.commit", false);
// 业务处理结束执行
try {
    // 这里进行提交
    // 用commitSync()提交由poll方法返回的最新偏移量,如果提交成功马上返回,提交失败则抛出异常。
    consumer.commitSync();
} catch (CommitFailedException e) {
    System.out.println(e.getMessage());
}

在于提交请求后,broker响应之前应用程序会一直阻塞,这样就会限制应用程序的吞吐量。
一旦发生再均衡,会增加重复消息的数量
异步提交(consumer.commitAsync(callback))

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zaji/5700057.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-12-17
下一篇 2022-12-17

发表评论

登录后才能评论

评论列表(0条)

保存