英文:
Spring kafka re-create Kafka Stream Topology in runtime
问题
我有一个基于Spring Boot、Spring Kafka和Kafka Streams的应用程序。
应用程序启动时，它会使用默认的主题列表创建Kafka Streams拓扑结构。
我需要在运行时编辑/重新创建拓扑结构。例如，当应用程序已经在运行时，会出现一个新的主题名称，我希望将此主题添加到我的拓扑中。
目前我在考虑以某种方式删除现有的拓扑结构，关闭和清理KafkaStreams，运行创建带有新主题名称的拓扑结构的逻辑，然后再次启动KafkaStreams。我不想重新启动我的应用程序。
有人可以建议我如何在运行时做到这一点吗？
英文:
I have an application that is based on spring boot, spring-kafka and kafka-streams.
When application starts up, it creates kafka streams topology with default list of topics.
What I need to do is edit/recreate topology in runtime. For example, when application already running, there is new topic name comes and I want to add this topic to my topology.
Currently I'm thinking about somehow to delete existing topology, close and clean up KafkaStreams, run logic where I create topology but with new topic name and start KafkaStreams again. I don`t want to restart my application.
Can someone suggest me how to do this in runtime?
答案1
得分: 1
我找到了1个解决方案。我扩展了StreamsBuilderFactoryBean：@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)@Primarypublic StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(KafkaStreamsConfiguration kStreamsConfigs) {return new DynamicStreamsBuilderFactoryBean(kStreamsConfigs);}public static class DynamicStreamsBuilderFactoryBean extends StreamsBuilderFactoryBean {private StreamsBuilder instance;public DynamicStreamsBuilderFactoryBean(final KafkaStreamsConfiguration streamsConfig) {super(streamsConfig);}@Overridepublic boolean isSingleton() {return false;}@Overrideprotected synchronized StreamsBuilder createInstance() {if (instance == null) {instance = new StreamsBuilder();}return instance;}@Overridepublic synchronized void stop() {instance = null;super.stop();}}而当我构建拓扑时，我不使用StreamsBuilder，而是使用StreamsBuilderFactoryBean#getObject()：@Componentpublic class DynamicStream {private final StreamsBuilderFactoryBean streamsBuilderFactoryBean;public void init() {StreamsBuilder builder = streamsBuilderFactoryBean.getObject();// 构建拓扑}// 当需要重新配置流时调用此方法public void reinitialize() {streamsBuilderFactoryBean.stop();init();streamsBuilderFactoryBean.start();}}
英文:
I found 1 solution.
I extend StreamsBuilderFactoryBean:
@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)@Primarypublic StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(KafkaStreamsConfiguration kStreamsConfigs) {return new DynamicStreamsBuilderFactoryBean(kStreamsConfigs);}public static class DynamicStreamsBuilderFactoryBean extends StreamsBuilderFactoryBean {private StreamsBuilder instance;public DynamicStreamsBuilderFactoryBean(final KafkaStreamsConfiguration streamsConfig) {super(streamsConfig);}@Overridepublic boolean isSingleton() {return false;}@Overrideprotected synchronized StreamsBuilder createInstance() {if (instance == null) {instance = new StreamsBuilder();}return instance;}@Overridepublic synchronized void stop() {instance = null;super.stop();}}
And when I build topology, I, instead of using StreamsBuilder, use StreamsBuilderFactoryBean#getObject():
@Component
public class DynamicStream {
private final StreamsBuilderFactoryBean streamsBuilderFactoryBean;public void init() {StreamsBuilder builder = streamsBuilderFactoryBean.getObject();//build topology}//call this method when stream reconfiguration is neededpublic void reinitialize() {streamsBuilderFactoryBean.stop();init();streamsBuilderFactoryBean.start();}
}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论