用java代码手动控制kafkaconsumer偏移量

Posted 和风细雨汪汪

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了用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

 

这里分享一个别人的源码分析:http://blog.csdn.net/chunlongyu/article/details/52663090>

原子操作( Atomic operations): 不可中断的一个或一系列操作,就像原子一样,不能再被拆分了,已经是最小单位了,当然在这里没有单位只有操作。 

 

以上是关于用java代码手动控制kafkaconsumer偏移量的主要内容,如果未能解决你的问题,请参考以下文章

Kafka Consumer多线程实例续篇

如何手动控制java中的事务

KafkaConsumer:`seekToEnd()` 不会让消费者从最新的偏移量消费

KafkaConsumer调用seek()方法报错

错题总结Java

KafkaConsumer 在 spark 中创建和删除 globalTempView 时对多线程访问不安全