首页 > 代码库 > 用java代码手动控制kafkaconsumer偏移量
用java代码手动控制kafkaconsumer偏移量
为应对消费结果需要存储到关系数据库中,避免数据库down时consumer继续消费的场景
http://kafka.apache.org
查了很多源码都记录下来,省的下次还要过滤源码。
- If the results of the consumption are being stored in a relational database, storing the offset in the database as well can allow committing both the results and offset in a single transaction. Thus either the transaction will succeed and the offset will be updated based on what was consumed or the result will not be stored and the offset won‘t be updated.
如果将结果存储在关系数据库中,那么在数据库中存储偏移量也可以允许在单个事务中提交结果和偏移量.。因此,要么事务成功,偏移量将根据所消耗的内容进行更新,否则结果将不会被存储,偏移量不会被更新.。
每个记录都有自己的偏移量,所以要管理你自己的偏移,你只需要做以下:
- Configure
enable.auto.commit=false
- Use the offset provided with each
ConsumerRecord
to save your position. - On restart restore the position of the consumer using
seek(TopicPartition, long)
.
这里分享一个别人的源码分析:http://blog.csdn.net/chunlongyu/article/details/52663090>
原子操作( Atomic operations): 不可中断的一个或一系列操作,就像原子一样,不能再被拆分了,已经是最小单位了,当然在这里没有单位只有操作。
用java代码手动控制kafkaconsumer偏移量
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。