kafka 同步生产者在第一次请求时需要更长的时间答案

【问题标题】：kafka sync producer takes longer time on the first requestkafka 同步生产者在第一次请求时需要更长的时间
【发布时间】：2017-08-07 14:57:08
【问题描述】：

我在 Spring Boot 微服务中使用 Spring Cloud Stream Kafka 同步生产者。每次我们部署服务时，对 kafka 的第一次调用需要 20 多秒才能将消息发布到 Topic。但所有后续调用几乎不需要 3 到 4 毫秒。这个问题也随机发生并且是间歇性的，但主要发生在我们重新启动服务时。我们正在使用 kafka 版本 0.9.0.1 和 gradle 依赖项，如下所示依赖{

compile('org.springframework.cloud:spring-cloud-starter-stream-kafka')
}
dependencyManagement {
imports {
mavenBom "org.springframework.cloud:spring-cloud-dependencies:Camden.SR3"
}
}

这里是应用程序。 yml

spring:
cloud:
stream:
bindings:
output:
content-type: application/json
destination: SOPOrderReceiveTopic
kafka:
binder:
brokers: "localhost:9092,localhost:9093"
headers: eventType
requiredAcks: -1
zkNodes: "localhost:2181"
bindings:
output:
producer:
configuration:
max:
block:
ms: 20000
reconnect:
backoff:
ms: 5000
request:
timeout:
ms: 30000
retries: 3
retry:
backoff:
ms: 10000
timeout:
ms: 30000
sync: true

我使用 org.springframework.cloud.stream.messaging.Source 作为输出通道，这是用于发布消息的方法

public void publish(Message event) {
    try {
        boolean result = source.output().send(event, orderEventConfig.getTimeoutMs());
        logger.log(LoggingEventType.INFORMATION, "MESSAGE SENT TO KAFKA : " + result);
    } catch (Exception publishingExceptionMessage) {
        logger.log(LoggingEventType.ERROR, "publish event to kafka failed!", publishingExceptionMessage);
        throw new PublishEventException("publish event to kafka failed for eventPayload: " + event.getPayload(),
                ThreadVariables.getTenantId());
    }
}

我知道同步生产者在性能方面较慢，因为它保证了消息的顺序和持久性，但为什么只有第一个请求需要这么长时间？这个问题是已知问题吗？它是否已在最新的 kafka 版本中修复。有人可以建议。谢谢

【问题讨论】：

20秒是否不包括实际应用启动时间？
是的，不包括服务开始时间

标签： spring-boot apache-kafka spring-cloud-stream spring-kafka

【解决方案1】：

使用以下依赖项下载的 Spring Cloud Stream 的版本似乎存在问题，

imports { 
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:Camden.SR3"
     }

尝试升级spring cloud stream并检查。它应该修复 springboot 服务启动后在 kafka 服务器上的第一次发布调用的延迟。

dependencies {  
    compile('org.springframework.cloud:spring-cloud-stream-binder-kafka')
}


     ext { springCloudVersion = 'Dalston.RELEASE' }

   dependencyManagement {   
        imports { 
                        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" 
                }
            }

【讨论】：