为什么async producer使用linger生成消息。ms和batch。大小设置为大值，自动刷新设置为false？

Question

我正在使用spring kafka 2.2.8，并使用以下设置编写一个简单的异步生成器：

producer config key : compression.type  and value is : none
producer config key : request.timeout.ms  and value is : 10000
producer config key : acks  and value is : all
producer config key : batch.size  and value is : 33554431
producer config key : delivery.timeout.ms  and value is : 1210500
producer config key : retry.backoff.ms  and value is : 3000
producer config key : key.serializer  and value is : class org.apache.kafka.common.serialization.StringSerializer
producer config key : security.protocol  and value is : SSL
producer config key : retries  and value is : 3
producer config key : value.serializer  and value is : class io.confluent.kafka.serializers.KafkaAvroSerializer
producer config key : max.in.flight.requests.per.connection  and value is : 1
producer config key : linger.ms  and value is : 1200000
producer config key : client.id  and value is : <<my app name>>

我使用下面的代码片段打印了上面的生产者设置：

DefaultKafkaProducerFactory defaultKafkaProducerFactory = (DefaultKafkaProducerFactory) mykafkaProducerFactory;
        Set<Entry> set  = defaultKafkaProducerFactory.getConfigurationProperties().entrySet();
        set.forEach( item ->
                System.out.println("producer config key : "+item.getKey()+"  and value is : "+item.getValue())
        );

现在我通过调用下面的构造函数来创建一个KafkaTemplate，其中autoFlush为false

public KafkaTemplate(mykafkaProducerFactory, boolean autoFlush) 

现在我有一个异步生产者在10秒内产生10条消息。然后，令人惊讶的是，我在几秒钟内就把这10条信息全部发布到了这个主题上，我确信这10条信息的总和要比我的批量小得多。尺码：33554431

现在我的问题是

1. 为什么要发布消息而不是等待linger.ms或batch.size才生成消息？

Answer 1

看起来您没有正确设置这些属性；展示你是如何设置它们的。我刚测试了一下

batch.size=1000000
linger.ms=10000

并背靠背发送了10条信息，它们花了整整10秒到达消费者手中。

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.type=batch

spring.kafka.producer.properties.batch.size=1000000
spring.kafka.producer.properties.linger.ms=10000

@SpringBootApplication
public class So62820095Application {


    private static final Logger LOG = LoggerFactory.getLogger(So62820095Application.class);


    public static void main(String[] args) {
        SpringApplication.run(So62820095Application.class, args);
    }

    @KafkaListener(id = "so62820095", topics = "so62820095")
    public void listen(List<String> in) {
        LOG.info(in.toString());
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so62820095").partitions(1).replicas(1).build();
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> IntStream.range(0,  10).forEach(i -> template.send("so62820095", "foo" + i));
    }

}