提高部署在 PCF 中的 Spring Boot 应用程序中 Kafka 侦听器的性能答案

【问题标题】：Improve performance for Kafka listeners in spring boot application deployed in PCF提高部署在 PCF 中的 Spring Boot 应用程序中 Kafka 侦听器的性能
【发布时间】：2020-11-14 03:43:31
【问题描述】：

我正在处理需要处理近 600 条消息/秒的用例（订阅主题、转换、保存到 SQL Server 表并返回主题），但我们每 5 个实例仅处理 100 条消息/秒。我们不能增加更多的实例来实现这一点。有什么建议会有帮助吗？

技术和基础设施：带有部署在 PCF 中的 Kafka 侦听器（无批处理侦听器）的 Spring Boot 应用程序。 source 和 out topic 每个都有 10 个分区。默认属性和设置正在使用。转换只需几毫秒。

【问题讨论】：

您需要对应用程序进行概要分析以查看瓶颈在哪里；可能是数据库？由于每个实例有 2 个分区，您可以尝试将并发性增加到 2，以便每个分区将在不同的线程上处理。

标签： java spring-boot spring-kafka pcf

【解决方案1】：

我有一个类似的用例，我通过以下配置向每个侦听器添加并发 (10) 并增加队列中的分区来提高性能

@Bean
public ThreadPoolTaskExecutor messageProcessorExecutor() {
    ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
    exec.setCorePoolSize(poolSize);
    exec.setMaxPoolSize(poolMaxSize);
    exec.setKeepAliveSeconds(keepAlive);
    return exec;
}

@Bean
public ConsumerFactory<String, Request> consumerFactory() {
    DefaultKafkaConsumerFactory<String, Request> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfigs());
    consumerFactory.setKeyDeserializer(new StringDeserializer());
    consumerFactory.setValueDeserializer(new JsonDeserializer<>(Request.class));
    return consumerFactory;
}

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Request>> kafkaListenerContainerFactory(
        ThreadPoolTaskExecutor messageProcessorExecutor,
        ConsumerFactory<String, Request> consumerFactory) {

    ConcurrentKafkaListenerContainerFactory<String, Request> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setConcurrency(10);
    factory.getContainerProperties().setPollTimeout(pollTimeout);
    factory.getContainerProperties().setConsumerTaskExecutor(messageProcessorExecutor);
    return factory;
}

private Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupID);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return props;
}

【讨论】：