Spring Boot微服务集成Kafka作为消费者启动异常问题咨询

在Spring Boot项目中集成Kafka时启动抛出异常，相关信息如下：

错误日志 #

2021-07-13T13:22:13.12+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.129+08:00","@version":"1","message":"Located property source: ConfigService {name='configService', propertySources=[MapPropertySource {name='configClient'}, MapPropertySource ]}","logger_name":"org.springframework.cloud.bootstrap.config.PropertySourceBootstrapConfiguration","thread_name":"main","level":"INFO"
2021-07-13T13:22:13.14+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.148+08:00","@version":"1","message":"'cloud' property source added","logger_name":"org.cloudfoundry.reconfiguration.CloudPropertySourceApplicationContextInitializer","thread_name":"main","level":"INFO","level_value":20000}
2021-07-13T13:22:13.14+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.149+08:00","@version":"1","message":"Reconfiguration enabled","logger_name":"org.cloudfoundry.reconfiguration.CloudServiceReconfigurationApplicationContextInitializer","thread_name":"main","level":"INFO","level_value":20000}
2021-07-13T13:22:13.15+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.156+08:00","@version":"1","message":"Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@57e03347: startup date [Tue Jul 13 13:22:13 SGT 2021]; parent: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@290dbf45","logger_name":"org.springframework.context.annotation.AnnotationConfigApplicationContext","thread_name":"main","level":"INFO","level_value":20000}
2021-07-13T13:22:13.19+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.193+08:00","@version":"1","message":"Skipping reconfiguration, already configured","logger_name":"org.cloudfoundry.reconfiguration.CloudServiceReconfigurationBeanFactoryPostProcessor","thread_name":"main","level":"INFO","level_value":20000}
2021-07-13T13:22:13.21+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.215+08:00","@version":"1","message":"JSR-330 'javax.inject.Inject' annotation found and supported for autowiring","logger_name":"org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor","thread_name":"main","level":"INFO","level_value":20000}
2021-07-13T13:22:13.41+0800 [APP/PROC/WEB/0] OUT {"@timestamp":"2021-07-13T13:22:13.410+08:00","@version":"1","message":"Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'binderHealthIndicator' defined in org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration$RabbitHealthIndicatorConfiguration: Unsatisfied dependency expressed through method 'binderHealthIndicator' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rabbitTemplate' defined in org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration$CloudProfile$CloudConnectors$UseCloudConnectors: Unsatisfied dependency expressed through method 'rabbitTemplate' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rabbitConnectionFactory' defined in org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration$CloudProfile$CloudConnectors$UseCloudConnectors: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.amqp.rabbit.connection.ConnectionFactory]: Factory method 'rabbitConnectionFactory' threw exception; nested exception is org.springframework.cloud.CloudException: No unique service matching interface org.springframework.amqp.rabbit.connection.ConnectionFactory found. Expected 1, found 0","logger_name":"org.springframework.context.annotation.AnnotationConfigApplicationContext","thread_name":"main","level":"WARN","level_value":30000}

pom.xml相关依赖 #

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
   <groupId>io.pivotal.spring.cloud</groupId>
   <artifactId>spring-cloud-services-starter-service-registry</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-spring-service-connector</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-cloudfoundry-connector</artifactId>
</dependency>

疑问 #

集成过程中遗漏了哪些配置或依赖？ #

报错核心原因是Spring Boot自动装配了RabbitMQ相关配置，尝试创建Rabbit连接工厂失败，和Kafka集成本身无关，可按以下步骤修复：

• 排除RabbitMQ自动装配类
  在项目启动类的@SpringBootApplication注解中添加排除规则，避免加载Rabbit相关自动配置：

@SpringBootApplication(exclude = {
    org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.class,
    org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration.class
})

• 显式指定默认Binder为Kafka
  在application配置文件中添加如下配置，避免框架自动探测Binder时优先匹配Rabbit相关配置：

# 配置默认使用kafka binder
spring.cloud.stream.default-binder=kafka
# 配置Kafka服务地址，按实际环境修改
spring.cloud.stream.kafka.binder.brokers=127.0.0.1:9092
# 若使用独立ZK管理Kafka可添加以下配置，高版本Kafka可省略
spring.cloud.stream.kafka.binder.zk-nodes=127.0.0.1:2181

• （可选）排除多余Rabbit依赖
  如果项目完全不需要RabbitMQ能力，可排查pom依赖树，排除所有隐式引入的RabbitMQ相关依赖，从根源上避免触发自动装配。
  如果部署在Cloud Foundry环境，需要确认已在CF平台创建Kafka服务实例，并且绑定到当前应用，否则Cloud连接器会找不到对应Kafka服务实例。

内容的提问来源于stack exchange，提问作者Liam