Kafka消费者调查行为

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;
          break;
       }
   }
}

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

提前感谢！！

Answer 1

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

对于您在这个问题中陈述的用例，您不必自己显式地提交偏移量。您只需将enable.auto.commit设置为true，并将auto.offset.reset设置为aresty，这样，当没有使用者group.id时（换句话说，当您第一次开始从分区读取时），它就会启动。一旦在Kafka中存储了group.id和一些使用者偏移量，并且如果Kafka使用者进程死亡，它将从上次提交的偏移量继续，因为这是默认行为，因为当使用者启动时，它将首先查找是否有任何提交的偏移量，如果有，将从上次提交的偏移量继续，并且auto.offset.reset不会起作用。