Spring kafka在运行时重新创建kafka流拓扑

Question

我有一个应用程序，是基于Spring启动，SpringKafka和Kafka流。当应用程序启动时，它会创建带有默认主题列表的kafka流拓扑。我需要做的是在运行时编辑/重新创建拓扑。例如，当应用程序已经运行时，有新的主题名称出现，我想将此主题添加到我的拓扑结构中。目前，我正在考虑以某种方式删除现有的拓扑，关闭并清理KafkaStreams，在创建拓扑但使用新主题名称的地方运行逻辑，并再次启动KafkaStreams。我不想重新启动我的应用程序。有人能建议我如何在运行时做到这一点吗？

Answer 1

我找到了一个解决方案。我扩展了StreamsBuilderFactoryBean：

@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)
@Primary
public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(KafkaStreamsConfiguration kStreamsConfigs) {
    return new DynamicStreamsBuilderFactoryBean(kStreamsConfigs);
}

public static class DynamicStreamsBuilderFactoryBean extends StreamsBuilderFactoryBean {

    private StreamsBuilder instance;

    public DynamicStreamsBuilderFactoryBean(final KafkaStreamsConfiguration streamsConfig) {
        super(streamsConfig);
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    @Override
    protected synchronized StreamsBuilder createInstance() {
        if (instance == null) {
            instance = new StreamsBuilder();
        }
        return instance;
    }

    @Override
    public synchronized void stop() {
        instance = null;
        super.stop();
    }
}

当我构建拓扑时，我不使用StreamsBuilder，而是使用StreamsBuilderFactoryBean#getObject（）：

@Component

公共类动态流{

private final StreamsBuilderFactoryBean streamsBuilderFactoryBean;

public void init() {
    StreamsBuilder builder = streamsBuilderFactoryBean.getObject();
        //build topology
}

//call this method when stream reconfiguration is needed
public void reinitialize() {
    streamsBuilderFactoryBean.stop();
    init();
    streamsBuilderFactoryBean.start();
}

}