卡夫卡消费者 - 民意调查行为

Question

关于KafkaConsumer（> = 0.9），我正面临一些严重的问题，试图为我的需求实施解决方案。

让我们假设我有一个函数必须只读取来自kafka主题的n条消息。

例如：getMsgs(5) - >在主题中获取接下来的5条kafka消息。

所以，我有一个看起来像这样的循环。用实际正确的参数编辑。在这种情况下，消费者的max.poll.records参数设置为1，因此实际循环只迭代一次。不同的消费者（其中一些消息通过许多消息迭代）共享一个抽象的父亲（这个），这就是为什么它以这种方式编码的原因。 numMss部分专门针对这个消费者。

for (boolean exit= false;!exit;)
{
   Records = consumer.poll(config.pollTime);
   for (Record r:records) {
       processRecord(r); //do my things
       numMss++;
       if (numMss==maximum) //maximum=5
          exit=true;
   }
}

考虑到这一点，问题是poll（）方法可以获得超过5条消息。例如，如果它获得10条消息，我的代码将永远忘记其他5条消息，因为Kafka会认为它们已经消耗掉了。

我尝试提交偏移但似乎不起作用：

    consumer.commitSync(Collections.singletonMap(partition,
    new OffsetAndMetadata(record.offset() + 1)));

即使使用偏移配置，每当我再次启动消费者时，它都不会从第6条消息开始（记住，我只想要5条消息），但是从第11条开始（因为第一次轮询消耗了10条消息）。

有没有解决方案呢，或者（最肯定的）我错过了什么？

提前致谢！！

Answer 1

您可以将max.poll.records设置为您喜欢的任何数字，这样您最多可以在每次轮询中获得那么多记录。

对于您在此问题中声明的用例，您不必自己明确提交偏移量。你可以设置enable.auto.commit到trueand将auto.offset.reset设置为earliest，这样当没有消费者group.id时它就会启动（换句话说，当你第一次开始从分区读取时）。一旦你有一个group.id和一些消费者偏移存储在Kafka中，并且你的Kafka消费者进程死亡，它将从最后一次提交的偏移继续，因为它是默认行为，因为当消费者启动时它将首先查找是否有任何承诺抵消，如果是，将继续从最后承诺的抵消和auto.offset.reset将不会启动。

Answer 2

将auto.offset.reset属性设置为“latest”。然后尝试使用，您将从提交的偏移量中获取消耗的记录。

或者您在民意调查之前使用consumer.seek（TopicPartition，offset）api。

Answer 3

您是否通过将enable.auto.commit设置为false来禁用自动提交。如果要手动提交偏移量，则需要禁用它。如果没有，下一次调用poll（）将自动提交从上一次poll（）收到的消息的最新偏移量。

Answer 4

从Kafka 0.9开始，auto.offset.reset参数名称已更改;

当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时（例如因为该数据已被删除）该怎么办：

earliest: automatically reset the offset to the earliest offset

latest: automatically reset the offset to the latest offset

none: throw exception to the consumer if no previous offset is found for the consumer's group

anything else: throw exception to the consumer.