如何制作可重启的生产者? [英] How to make restart-able producer?
问题描述
kafka的最新版本支持完全一次语义(EoS).为了支持此概念,将额外的详细信息添加到每条消息.这意味着对您的消费者而言;如果您打印邮件的偏移量,它们不一定是顺序的.这使得轮询一个主题以阅读最后提交的消息变得更加困难.
Latest version of kafka support exactly-once-semantics (EoS). To support this notion, extra details are added to each message. This means that at your consumer; if you print offsets of messages they won't be necessarily sequential. This makes harder to poll a topic to read the last committed message.
在我的情况下,消费者打印出这样的内容
In my case, consumer printed something like this
Offset-0 0
Offset-2 1
Offset-4 2
问题:为了编写可重启的骄傲者;我轮询主题并阅读最后一条消息的内容.在这种情况下;最后一条消息将是offset#5,这不是有效的消费者记录.因此,我在代码中看到错误.
Problem: In order to write restart-able proudcer; I poll the topic and read the content of last message. In this case; last message would be offset#5 which is not a valid consumer record. Hence, I see errors in my code.
我可以使用以下网址提供的解决方案:获取最后一个邮件发送到kafka主题.唯一的问题是,而不是使用consumer.seek(partition,last_offset = 1);我会使用Consumer.seek(partition,last_offset-2).这样可以立即解决我的问题,但这不是理想的解决方案.
I can use the solution provided at : Getting the last message sent to a kafka topic. The only problem is that instead of using consumer.seek(partition, last_offset=1); I would use consumer.seek(partition, last_offset-2). This can immediately resolve my issue, but it's not an ideal solution.
对于用Java编写的消费者而言,获取上一次提交的消息的最可靠,最佳的解决方案是什么?或
是否可以使用用于分区的本地状态存储?或
Is it possible to use local state-store for a partition? OR
最推荐的存储最后一条消息以防止生产者失败的方法是什么?或
kafka连接器是否可以重新启动?我可以使用任何特定的API使生产者重新启动吗?
仅供参考-我不是在寻找快速解决方法
FYI- I am not looking for quick fix
推荐答案
就我而言,多个生产者将数据推送到一个大主题.因此,阅读整个主题将是一场噩梦.
In my case, multiple producers push data to one big topic. Therefore, reading entire topic would be nightmare.
我找到的解决方案是维护另一个主题,即"P1_Track",生产者可以在其中存储元数据.在交易中,制作人会将数据发送到一个大主题和P1_Track.
The solution that I found is to maintain another topic i.e. "P1_Track" where producer can store metadata. Within a transaction a producer will send data to one big topic and P1_Track.
当我重新启动制作器时,它将读取P1_Track并找出从哪里开始.
When I restart a producer, it will read P1_Track and figure out where to start from.
考虑将最后提交的消息存储在数据库中,并在生产者进程重新启动时使用它.
Thinking about storing last committed message in a database and using it when producer process restarts.
这篇关于如何制作可重启的生产者?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!