手动启动 Kafka Streams 处理器
Spring Cloud Stream Kafka Streams binder 在 Spring for Apache Kafka 的 StreamsBuilderFactoryBean 之上提供了一个名为 StreamsBuilderFactoryManager 的抽象层。该管理器 API 用于在基于 binder 的应用程序中控制每个处理器的多个 StreamsBuilderFactoryBean。因此,在使用 binder 时,如果您希望手动控制应用程序中各个 StreamsBuilderFactoryBean 对象的自动启动,就需要使用 StreamsBuilderFactoryManager。您可以使用属性 spring.kafka.streams.auto-startup 并将其设置为 false 来关闭处理器的自动启动。然后,在应用程序中,您可以使用类似以下代码的方式,通过 StreamsBuilderFactoryManager 来启动处理器。
@Bean
public ApplicationRunner runner(StreamsBuilderFactoryManager sbfm) {
return args -> {
sbfm.start();
};
}
当您希望应用程序在主线程中启动,并让 Kafka Streams 处理器单独启动时,此功能非常方便。例如,当您有一个需要恢复的大型状态存储时,如果处理器像默认情况那样正常启动,可能会阻塞您的应用程序启动。如果您正在使用某种活跃度探测机制(例如在 Kubernetes 上),它可能会认为应用程序已关闭并尝试重新启动。为了纠正这种情况,您可以将 spring.kafka.streams.auto-startup 设置为 false,并遵循上述方法。
请注意,在使用 Spring Cloud Stream binder 时,您并不是直接处理来自 Spring for Apache Kafka 的 StreamsBuilderFactoryBean,而是处理 StreamsBuilderFactoryManager,因为 StreamsBuilderFactoryBean 对象是由 binder 内部管理的。