手动启动Kafka Streams处理器

Spring Cloud Stream Kafka Streams 绑定器提供了一个抽象功能,称为StreamsBuilderFactoryManagerStreamsBuilderFactoryBean来自《阿帕奇·卡夫卡》的春季。 该管理器API用于控制多重StreamsBuilderFactoryBean在基于活页夹的应用程序中,每个处理器的计算。 因此,使用绑定器时,如果你想手动控制各种自动启动StreamsBuilderFactoryBean应用中的对象,你需要使用StreamsBuilderFactoryManager. 你可以使用这块地产Spring.kafka.streams.auto-startup并将此设为false为了关闭处理器的自动启动。 然后,在应用程序中,你可以用以下方法启动处理器StreamsBuilderFactoryManager.spring-doc.cadn.net.cn

@Bean
public ApplicationRunner runner(StreamsBuilderFactoryManager sbfm) {
    return args -> {
        sbfm.start();
    };
}

这个功能很方便,当你希望应用从主线程启动,并让 Kafka Streams 处理器单独启动时。 例如,当你需要恢复一个大型状态存储时,如果处理器像默认情况下一样正常启动,可能会导致你的应用程序无法启动。 如果你使用某种活体探测机制(例如在 Kubernetes 上),它可能会认为应用程序宕机并尝试重启。 为了纠正这个问题,你可以设置Spring.kafka.streams.auto-startupfalse并按照上面的方法进行。spring-doc.cadn.net.cn

请记住,使用Spring Cloud Stream绑定器时,你并不是直接处理StreamsBuilderFactoryBean更像是《阿帕奇·卡夫卡的Spring》StreamsBuilderFactoryManager,作为StreamsBuilderFactoryBean对象由绑定器内部管理。spring-doc.cadn.net.cn