如何实现Spring Kafka消费者的高性能

Question

如何提高Kafka消费者的绩效？我有（并且需要）至少一次Kafka消费语义学

我有以下配置。processInDB（）需要2分钟才能完成。因此，仅处理10条消息（全部在单个分区中）就需要20分钟（假设每条消息2分钟）。我可以在不同的线程中调用processInDB，但我可能会丢失消息！。如何在2到4分钟的时间窗口内处理所有10条消息？

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "grpid-mytopic120112141");
  props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
  props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");

ConcurrentKafkaListenerContainerFactory<String, ValidatedConsumerClass> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(AckMode.RECORD);
        factory.setErrorHandler(errorHandler());

下面是我的Kafka消费者代码。

@KafkaListener(id = "foo", topics = "mytopic-3", concurrency = "6", groupId = "mytopic-1-groupid")
    public void consumeFromTopic1(@Payload @Valid ValidatedConsumerClass message, ConsumerRecordMetadata c)  {

        dbservice.processInDB(message);


    }

Answer 1

另一个不完全与spring kafka相关的建议，正如您在标签中所述，您也在探索消费者api，而不仅仅是spring kafka，因此我允许自己在这里提出建议，您可能希望测试此api

https://www.confluent.io/blog/introducing-confluent-parallel-message-processing-client/

https://github.com/confluentinc/parallel-consumer

• 其处于阿尔法阶段，因此不建议用于生产，但可能也会注意这一点

但正如我在前面的评论中所述，您可能只想创建更多的分区

Answer 2

使用批处理侦听器会有所帮助—您只需在侦听器中保留使用者线程，直到所有单个记录都完成处理。

在下一个版本（今天发布的2.8.0-M1里程碑）中，支持无序的手动确认，其中框架将提交推迟到“填补空白”为止https://docs.spring.io/spring-kafka/docs/2.8.0-M1/reference/html/#x28-ooo提交