SpringKafka抵消promise方式

Question

如果我不想使用自动通信模式，sping提供了另一种方法。

spring kafkfa/#提交补偿为我们提供了以下关于提交补偿的信息：

RECORD-当侦听器在处理记录后返回时提交偏移量
BATCH-处理poll（）返回的所有记录后提交偏移量
TIME-当poll（）返回的所有记录都已被处理时，只要超过自上次提交以来的确认时间，就提交偏移量
COUNT-当poll（）返回的所有记录都已被处理时，只要自上次提交以来已收到ackCount记录，就提交偏移量
COUNT\u TIME-类似于TIME和COUNT，但如果任一条件为真，则执行提交
手册-消息侦听器负责确认（）；然后，应用与批处理相同的语义
MANUAL_IMMEDIATE-确认后立即提交偏移量。侦听器调用acknowledge（）方法。

我有几个问题：

TIME据我所知，Spring框架中的某个地方存在循环

while(true){
   data = consumer.poll();
   data.foreach(record->listener.listen(record))
}

民意测验多久进行一次？

时间是提交偏移量的唯一标准吗？假设投票返回了100条记录，当ackTime过期时-只处理了60条记录？

我没有抓住MANUAL_IMMEDIATE和MANUAL之间的区别

请为我澄清这些问题。

附笔。

据我所知，Garry Russel foreach的回答如下：

while(true){
   data = consumer.poll();
   data.foreach(record->new Thread(()->listener.listen(record)).start());
}

Answer 1

这取决于版本；最近的1.3版本有一个更简单的线程模型，由KIP-62提供了便利。

使用该版本，在调用者线程上调用侦听器；直到所有当前记录都被消耗掉，下一次轮询才会发生。除了RECORD（和MANUAL*）之外，提交的决定是在所有记录都发送给侦听器之后确定的。

MANUAL_IMMEDIATE就是这个意思；偏移量在用户处理时立即提交；使用MANUAL，手动偏移量在所有记录发送给侦听器后提交。

与早期版本相比要复杂一些；可以提取一个或两个额外的批，并且在每次轮询之前执行ack，以便在将第一批中的所有记录发送到侦听器之前提交偏移量。

编辑

在下面回复您的评论...

是的；线程在1.3中更改。在此之前，我们必须继续轮询消费者以避免代理重新平衡分区。与

所以，最坏的情况是容器包含3组记录——侦听器当前正在处理的记录、队列中的记录和我们无法放入队列的记录。任何未完成的偏移提交（手动或其他）都会在每次轮询之前执行。

轮询线程会等待处理器线程吗？

不我们不能这样做，因为这会导致重新平衡——就像我们在消费者线程上调用侦听器一样。

KIP-62实际上在0.10.1.0客户端中得到了修复，但直到1.3版本，我们才更改线程；多亏了KIP-62，这是一个重大的简化，我建议使用该版本。