StreamsBuilderFactoryBean 配置器

DeepSeek V3 中英对照 StreamsBuilderFactoryBean configurer

通常需要自定义创建 KafkaStreams 对象的 StreamsBuilderFactoryBean。基于 Spring Kafka 提供的底层支持，绑定器允许您自定义 StreamsBuilderFactoryBean。您可以使用 StreamsBuilderFactoryBeanConfigurer 来自定义 StreamsBuilderFactoryBean 本身。然后，一旦通过此配置器访问到 StreamsBuilderFactoryBean，您就可以使用 KafkaStreamsCustomzier 来自定义相应的 KafkaStreams。这两个自定义器都是 Spring for Apache Kafka 项目的一部分。

以下是使用 StreamsBuilderFactoryBeanConfigurer 的示例。

@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return sfb -> sfb.setStateListener((newState, oldState) -> {
         //Do some action here!
    });
}

以上展示了自定义 StreamsBuilderFactoryBean 的示例操作。实际上，你可以调用 StreamsBuilderFactoryBean 上任何可用的修改操作来自定义它。此自定义器将在工厂 bean 启动前由绑定器调用。

一旦你获得了对 StreamsBuilderFactoryBean 的访问权限，你也可以自定义底层的 KafkaStreams 对象。以下是一个实现此操作的蓝图。

@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return factoryBean -> {
        factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
            @Override
            public void customize(KafkaStreams kafkaStreams) {
                kafkaStreams.setUncaughtExceptionHandler((t, e) -> {

                });
            }
        });
    };
}

KafkaStreamsCustomizer 将在底层的 KafkaStreams 启动之前由 StreamsBuilderFactoryBean 调用。

整个应用中只能有一个 StreamsBuilderFactoryBeanConfigurer。那么，我们如何处理多个 Kafka Streams 处理器，因为每个处理器都由独立的 StreamsBuilderFactoryBean 对象支持？在这种情况下，如果这些处理器需要不同的自定义配置，那么应用程序需要根据应用程序 ID 应用一些过滤器。

例如，

@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return factoryBean -> {
        if (factoryBean.getStreamsConfiguration().getProperty(StreamsConfig.APPLICATION_ID_CONFIG)
                .equals("processor1-application-id")) {
            factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
                @Override
                public void customize(KafkaStreams kafkaStreams) {
                    kafkaStreams.setUncaughtExceptionHandler((t, e) -> {

                    });
                }
            });
        }
    };

使用 StreamsBuilderFactoryBeanConfigurer 注册全局状态存储

如上所述，binder 并未提供将全局状态存储注册为功能的一流方法。为此，你需要通过 StreamsBuilderFactoryBeanConfigurer 使用自定义器。具体操作方式如下。

@Bean
public StreamsBuilderFactoryBeanConfigurer customizer() {
    return streamsBuilderFactoryBean -> {
        try {
            streamsBuilderFactoryBean.setInfrastructureCustomizer(new KafkaStreamsInfrastructureCustomizer() {
                  @Override
                  public void configureBuilder(StreamsBuilder builder) {
                      builder.addGlobalStore(
                              ...
                      );
                  }
              });
        }
        catch (Exception e) {

        }
    };
}

对 StreamsBuilder 的任何自定义都必须通过 KafkaStreamsInfrastructureCustomizer 来完成，如上所示。如果调用 StreamsBuilderFactoryBean#getObject() 来获取 StreamsBuilder 对象，这可能无法正常工作，因为该 Bean 可能仍在初始化中，从而引发一些循环依赖问题。

如果你有多个处理器，你需要通过使用上面提到的应用程序ID过滤掉其他StreamsBuilderFactoryBean对象，将全局状态存储附加到正确的StreamsBuilder上。

使用 StreamsBuilderFactoryBeanConfigurer 注册生产异常处理器

在错误处理部分，我们指出绑定器并未提供处理生产异常的一流方法。尽管如此，您仍可使用 StreamsBuilderFactoryBean 自定义器来注册生产异常处理器。请参阅下文。

@Bean
public StreamsBuilderFactoryBeanConfigurer configurer() {
    return fb -> {
        fb.getStreamsConfiguration().put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
                            CustomProductionExceptionHandler.class);
    };
}

再次强调，如果你有多个处理器，可能需要针对正确的 StreamsBuilderFactoryBean 进行适当设置。你也可以通过配置属性添加此类生产异常处理器（更多相关内容见下文），但如果你选择采用编程方式，这也是一个可选项。