代码之家 › 专栏 › 技术社区 › kan

卡夫卡消费者偏移自动复位等参数

kafka-consumer-api apache-kafka java

kan · 技术社区 · 3 年前

我有一个KafkaConsumer,它需要订阅两个主题 topicA 和 topicB 。但是,我需要一些不同的参数。例如,如果我需要 auto.offset.reset 对于 topicA 是 earilest 而对于 主题B 应该是 latest 。我看不出有什么简单的方法可以做到这一点。一种选择是运行两个使用者,但在这种情况下,我需要为它们提供两个轮询线程,因此应该处理多线程。有什么更简单的方法吗?

0 回复 | 直到 3 年前

OneCricketeer Gabriele Mariotti 3 年前

创建两个(或多个)线程是正确的。

使用者不是线程安全的,无论如何都应该与其他进程隔离和分离。

您可以使用更高级别的Kafka库(例如Vert.x/Spring)来简化这一过程。

sawim 3 年前

如果你需要使用一个消费者,那么我认为你可以设置 auto.reset.offset 到 latest 然后移动的偏移量 topicA 手动(如果需要)。为此,在订阅和轮询循环之间,您可以:

获取分配给使用者的分区(方法 assignment )
检查主题的承诺偏移量 topicA 通过筛选的分区 topicA 从上一点(方法 committed ). 如果结果是 null ,然后打电话 seekToBeginning 方法

推荐文章

MrSham · 如何确保旧版本0.10.0在卡夫卡中只处理一次

3 年前

BnJ · 如何知道卡夫卡何时提交记录?

7 年前

addictedtohaskell · 为什么卡夫卡消费者无视我在汽车上的“最早”指令。抵消重置参数,从而不从绝对第一个事件读取我的主题?

7 年前

maverick · 消息预处理(主题-主题)-Kafka Connect API vs.Streams vs Kafka Consumer?

7 年前

Renato Garcia · 如何设置隔离。等级卡夫卡骆驼组件

7 年前

Javier · 使用kafka主题中的json值,并使用JQ将其写入csv文件并格式化

7 年前

Karol Król · 如何在一个稳定的卡夫卡消费群体中转移话题的偏移量?

7 年前

maxness · Spark streaming:错误StreamingContext:无法构造kafka使用者

7 年前

JJ15k · 卡夫卡分区分配协议

7 年前

Jary zhen · 日志使用kafka 0.11.0调试(“组{}的协调器发现失败,刷新元数据”,groupId)。x个

7 年前