Spring Kafka-动态创建流

Question

我需要从配置文件动态创建kafka流，其中包含每个流的源主题名称和配置。应用程序需要有几十个Kafka流和流将是不同的每个环境（例如阶段，prod）。它可能做到这一点与sping-kafka库？

我们可以通过Kafka流轻松做到这一点：

@Bean
public List<KafkaStreams> kafkaStreams() {
    return streamRouteProperties.stream()
            .map(routeProperty -> createKafkaStream(routeProperty))
            .collect(toList());
}

private KafkaStreams createKafkaStream(KafkaConfigurationProperties kafkaProperties) {
    StreamsBuilder builder = new StreamsBuilder();
    KStream<Object, String> stream = builder.stream(kafkaProperties.getTopicName());
    Topology topology = builder.build();
    StreamsConfig streamsConfig = new StreamsConfig(kafkaProperties.getSettings());
    return new KafkaStreams(topology, streamsConfig);
}

我们需要实现springSmartLifecycle接口，这样所有流都将自动启动和关闭。

是否可以使用sping-kafka做同样的事情？正如我所看到的，我们需要在代码中创建每个Kafka流，我看不到如何使用StreamsBuilderFactoryBean创建Kafka流列表的可能性。

@Bean
public KStream<?, ?> kStream(StreamsBuilder streamsBuilder) {
    Consumed<String, String> consumed = ..;
    KStream<String, String> kStream = streamsBuilder.stream(topicName, consumed);
    kStream.process(() -> eventProcessor);
    return kStream;   
}

@Bean
public FactoryBean<StreamsBuilder> streamsBuilder() {
    return new StreamsBuilderFactoryBean(streamsConfig);
}

但是如何使用StreamsBuilderFactoryBean动态创建Kafka流列表呢？

Answer 1

StreamsBuilderFactoryBean只是为Spring应用程序上下文带来了一些自以为是的、方便的API，但这并不意味着您应该始终与之捆绑在一起。

幸运的是，StreamsBuilderFactoryBean与常规Kafka流相比没有太大的价值。它的最大功能是对内部创建的KafkaStreams进行lfecycle控制。

您可以随意使用原始的Kafka Streams API，不要试图将代码过于复杂化，使其成为基于StreamsBuilderFactoryBean的您的需求，StreamsBuilderFactoryBean实际上是为一组静态选项设计的。