如何在spring boot中配置两个Kafka StreamsBuilderFactoryBean实例 [英] how to configure two instances of Kafka StreamsBuilderFactoryBean in spring boot

问题描述

使用 spring-boot-2.1.3、spring-kafka-2.2.4，我想有两个流配置(例如有不同的 application.id，或连接到不同的集群等).所以我几乎根据文档定义了第一个流配置，然后添加了第二个，具有不同的名称，以及第二个 StreamsBuilderFactoryBean(也具有不同的名称):

Using spring-boot-2.1.3, spring-kafka-2.2.4, I want to have two streams configurations (e.g. to have different application.id, or connect to different cluster, etc). So I defined the first stream configuration pretty much according to the docs, then added a second one, with a different name, and a second StreamsBuilderFactoryBean (also with a different name):

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "myappId1000");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    //...
    return new KafkaStreamsConfiguration(props);
}

@Bean(name = "myKappConfig")
public KafkaStreamsConfiguration myKafkaAppIdConfiguration() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "myappId9999");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    //...
    return new KafkaStreamsConfiguration(props);
}

@Bean(name = "myKappConfigStreamBuilder")
public StreamsBuilderFactoryBean myAppStreamBuilder(
        @Qualifier("myKappConfig") KafkaStreamsConfiguration myKafkaAppIdConfiguration) {
    return new StreamsBuilderFactoryBean(myKafkaAppIdConfiguration);
}

但是，当我尝试运行该应用程序时，我得到:

However, when I try to run the app, I get:

kafkaStreamsFactoryBeanConfigurer 方法中的参数 0org.springframework.boot.autoconfigure.kafka.KafkaStreamsAnnotationDrivenConfiguration需要一个 bean，但找到了 2 个:- &defaultKafkaStreamsBuilder:由类路径资源中的defaultKafkaStreamsBuilder"方法定义[org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.class]- &myKappConfigStreamBuilder:由类路径资源中的myAppStreamBuilder"方法定义[com/teramedica/kafakaex001web/KafkaConfig.class]

Parameter 0 of method kafkaStreamsFactoryBeanConfigurer in org.springframework.boot.autoconfigure.kafka.KafkaStreamsAnnotationDrivenConfiguration required a single bean, but 2 were found: - &defaultKafkaStreamsBuilder: defined by method 'defaultKafkaStreamsBuilder' in class path resource [org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.class] - &myKappConfigStreamBuilder: defined by method 'myAppStreamBuilder' in class path resource [com/teramedica/kafakaex001web/KafkaConfig.class]

因为 spring-boot 自动配置中的代码:

because the code in the spring-boot autoconfigure does:

@Bean
public KafkaStreamsFactoryBeanConfigurer kafkaStreamsFactoryBeanConfigurer(
        StreamsBuilderFactoryBean factoryBean) {
    return new KafkaStreamsFactoryBeanConfigurer(this.properties, factoryBean);
}

除了完全替换 KafkaStreamsAnnotationDrivenConfiguration，我如何定义多个 StreamsBuilderFactoryBean.或者，我如何更改给定流的属性?

Short of replacing the KafkaStreamsAnnotationDrivenConfiguration entirely, how do I define more than one StreamsBuilderFactoryBean. Or alternately, how can I change the properties for a given stream?

推荐答案

用 @Primary 标记一个工厂 bean.

Mark one factory bean with @Primary.

这篇关于如何在spring boot中配置两个Kafka StreamsBuilderFactoryBean实例的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！