Spring Kafka轮询，将@kafkalistener和listener ack-mode设置为记录

Question

    ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(30);
        factory.getContainerProperties().setSyncCommits(true);
        return factory;
    }
    @KafkaListener(topics = "topic1", containerFactory="kafkaListenerContainerFactory")
    public void handleMessage(final ConsumerRecord<Object, String> arg0) throws Exception {
    }
    @KafkaListener(topics = "topic2", containerFactory="kafkaListenerContainerFactory")
    public void handleMessage(final ConsumerRecord<Object, String> arg0) throws Exception {
    }
    @KafkaListener(topics = "topic3", containerFactory="kafkaListenerContainerFactory")
    public void handleMessage(final ConsumerRecord<Object, String> arg0) throws Exception {
    }

my Listener.AckMode为return并Enable.Auto.Commit设置为false，Partition.Assignment.Strategy：org.apache.kafka.clients.consumer.RoundroBinAssignor

1）我对并发的理解是，由于我将并发（在工厂级别）设置为30，并且我总共有30个分区（用于所有三个主题）可供读取，每个线程将被分配一个分区。我的理解正确吗？如果我在@KafKalistener注释中再次覆盖并发，会有什么影响？

2）当spring调用poll（）方法时，它是否从所有三个主题进行轮询？

3）由于我将listener.ackmode设置为return，它是否会等到在一个poll（）中返回的所有记录都完成后再发出下一个poll（）？如果我的记录处理的时间比max.poll.interval.ms长，又会发生什么？假设在一个poll（）调用中返回1-100个偏移量，而我的代码在max.poll.interval.ms被命中之前只能处理50个偏移量，此时spring会发出另一个poll吗，因为它已经命中max.poll.interval.ms？如果是的话，下一个poll（）会从偏移量51返回记录吗？

真的很感谢你的时间和帮助

Answer 1

我的Listener.AckMode已返回

没有这样的ACKMODE；由于您没有在工厂中设置它，所以您的实际ack模式是批处理（默认）。要使用ack模式记录（如果您的意思是这样的话），您必须这样配置工厂容器属性。

我对并发的理解是...

spring会不会在这个时候再发布一次民意调查

Spring无法控制--如果花费的时间太长，使用者线程就在侦听器中--使用者不是线程安全的，所以我们不能发出异步轮询。

必须在max.poll.interval.ms内处理max.poll.records以避免Kafka重新平衡分区。