spring-cloud-stream - 无法使用 Spring Cloud Stream 在消费者端设置多个绑定到同一目的地和组
问题描述
我打算为 Kafka Consumer 弹性设置多个绑定。更具体地说,备用侦听器与主侦听器期望代理 IP 具有相同的目标和组。备份侦听器的 autoStartUp 在启动时关闭,但会在发生故障转移时以编程方式打开。
但是,微服务在启动时会抛出以下异常:
org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer:
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:461) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:90) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:143) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.doBindConsumer(BindingService.java:169) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:126) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.AbstractBindableProxyFactory.createAndBindInputs(AbstractBindableProxyFactory.java:112) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.doStartWithBindable(InputBindingLifecycle.java:58) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:608) ~[na:1.8.0_181]
at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:57) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:34) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:182) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:53) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:360) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:158) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:122) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:894) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:162) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:553) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:747) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at com.example.demo.DemoApplication.main(DemoApplication.java:14) ~[classes/:na]
Caused by: org.springframework.beans.factory.support.BeanDefinitionOverrideException: Invalid bean definition with name 'my-kafka-topic-name.my-kafka-consumer-group-name.errors.recoverer' defined in null: Cannot register bean definition [Root bean: class [org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer]; scope=; abstract=false; lazyInit=null; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=null; factoryMethodName=null; initMethodName=null; destroyMethodName=null] for bean 'my-kafka-topic-name.my-kafka-consumer-group-name.errors.recoverer.errors.recoverer': There is already [Root bean: class [org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer]; scope=; abstract=false; lazyInit=null; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=null; factoryMethodName=null; initMethodName=null; destroyMethodName=null] bound.
at org.springframework.beans.factory.support.DefaultListableBeanFactory.registerBeanDefinition(DefaultListableBeanFactory.java:927) ~[spring-beans-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.GenericApplicationContext.registerBeanDefinition(GenericApplicationContext.java:323) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.GenericApplicationContext.registerBean(GenericApplicationContext.java:471) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:681) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:633) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:644) ~[spring-cloud-stream-binder-kafka-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:148) ~[spring-cloud-stream-binder-kafka-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:407) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
... 24 common frames omitted
配置:
spring:
cloud:
function:
definition: mainListener;backupListener;
stream:
default-binder: my-main-binder
function:
bindings:
mainListener-in-0: event-in
backupListener-in-0: backup-event-in
bindings:
event-in: <-- main listener
binder: my-main-binder
consumer:
qulifer: local
use-native-decoding: true
destination: my-kafka-topic-name
group: my-kafka-consumer-group-name
backup-event-in: <-- backup listener with autoStartup turned off
binder: my-backup-binder
consumer:
autoStartup: false
use-native-decoding: true
destination: my-kafka-topic-name <-- same destination and group as the main listener
group: my-kafka-consumer-group-name
binders:
my-main-binder:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
brokers: ip_1; ip_2;
my-backup-binder:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
brokers: backup_ip_1; backup_ip_2;
依赖项:spring-cloud-function-context:3.0.2.RELEASE spring-cloud-stream-binder-kafka:3.0.2.RELEASE
另一个帖子的接受答案表明将添加 Qualifier 属性来解决此问题。但是我在 spring-cloud-stream:3.0.2.RELEASE 的 org.springframework.cloud.stream.binder.ConsumerProperties 类中找不到它
启动消费者时抛出异常 - (无法为微服务中的不同通道分配相同的组名)
https://github.com/garyrussell/spring-cloud-stream/commit/5b87b8cae494ae9568d924f64adc436374a67ea7
更新# 1: 按照建议将 allow-bean-definition-overriding 配置为 true 后,我看到了一个不同的异常,如下所示:
org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer:
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:461) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:90) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:143) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.doBindConsumer(BindingService.java:169) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:126) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.AbstractBindableProxyFactory.createAndBindInputs(AbstractBindableProxyFactory.java:112) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.doStartWithBindable(InputBindingLifecycle.java:58) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:608) ~[na:1.8.0_181]
at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:57) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:34) [spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:182) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:53) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:360) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:158) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:122) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:894) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:162) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:553) ~[spring-context-5.2.3.RELEASE.jar:5.2.3.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:747) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) ~[spring-boot-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at com.example.demo.DemoApplication.main(DemoApplication.java:14) ~[classes/:na]
Caused by: java.lang.IllegalStateException: Only one LastSubscriberMessageHandler is allowed
at org.springframework.cloud.stream.binder.BinderErrorChannel.subscribe(BinderErrorChannel.java:44) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:712) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:633) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:644) ~[spring-cloud-stream-binder-kafka-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:148) ~[spring-cloud-stream-binder-kafka-3.0.2.RELEASE.jar:3.0.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:407) ~[spring-cloud-stream-3.0.2.RELEASE.jar:3.0.2.RELEASE]
... 24 common frames omitted
解决方案
目前默认不支持该配置;错误基础设施 bean 名称基于组和目标。
如果您不介意两个消费者使用相同的错误基础架构(在这种情况下听起来不错),您可以设置
spring.main.allow-bean-definition-overriding=true
推荐阅读
- vue.js - snabdom、hyperscript 和 Vue 2 有什么关系?
- java - 我的“/” requestMapping 向我发送了 WhiteLabel“/”错误
- javascript - 在选择时显示选定的下拉项
- pandas - Spark 与 Scala 和 Pandas
- python - 我在这里想念什么?python中的简单for循环
- html - 如何从列表中创建新的 HtmlAgilityPack.HtmlDocument
从第一次“选择/过滤”运行? - python - 从一列 csv 文件创建数据集,其中数据用空格分隔
- ssh - 连接到 CentOS 7.9 时出现 SSH 错误权限被拒绝
- binary - 如果给出另一个字母的二进制代码,如何找到给定字母的 ascii 代码?
- c++ - 如何在 C 中循环 if 语句