如何制作可重启的生产者? [英] 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屋！