|
对于最新的稳定版本，请使用 spring-cloud-stream 4.3.0！ |
编程模型的辅助工具
单个应用程序中的多个 Kafka Streams 处理器
Binder 允许在单个 Spring Cloud Stream 应用程序中拥有多个 Kafka Streams 处理器。 您可以申请如下。
@Bean
public java.util.function.Function<KStream<Object, String>, KStream<Object, String>> process() {
...
}
@Bean
public java.util.function.Consumer<KStream<Object, String>> anotherProcess() {
...
}
@Bean
public java.util.function.BiFunction<KStream<Object, String>, KTable<Integer, String>, KStream<Object, String>> yetAnotherProcess() {
...
}在这种情况下，绑定器将创建 3 个具有不同应用程序 ID 的单独 Kafka Streams 对象（更多内容见下文）。 但是，如果应用程序中有多个处理器，则必须告诉 Spring Cloud Stream 需要激活哪些功能。 以下是激活这些功能的方法。
spring.cloud.function.definition: process;anotherProcess;yetAnotherProcess
如果您不希望立即激活某些功能，可以将其从此列表中删除。
当您拥有单个 Kafka Streams 处理器和其他类型的Function同一应用程序中的 bean，通过不同的绑定器处理（例如，基于常规 Kafka 消息通道绑定器的函数 bean）
Kafka Streams 应用程序 ID
应用程序 ID 是您需要为 Kafka Streams 应用程序提供的必填属性。 Spring Cloud Stream Kafka Streams 绑定器允许您以多种方式配置此应用程序 ID。
如果应用程序中只有一个处理器，则可以使用以下属性在绑定器级别进行设置：
spring.cloud.stream.kafka.streams.binder.applicationId.
为了方便起见，如果您只有一个处理器，您还可以使用spring.application.name作为属性来委托应用程序 ID。
如果应用程序中有多个 Kafka Streams 处理器，则需要为每个处理器设置应用程序 ID。 对于函数模型，您可以将其作为属性附加到每个函数。
例如，假设您有以下功能。
@Bean
public java.util.function.Consumer<KStream<Object, String>> process() {
...
}和
@Bean
public java.util.function.Consumer<KStream<Object, String>> anotherProcess() {
...
}然后，您可以使用以下活页夹级别属性为每个应用程序设置应用程序 ID。
spring.cloud.stream.kafka.streams.binder.functions.process.applicationId
和
spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.applicationId
对于基于函数的模型，这种在绑定级别设置应用程序 ID 的方法也有效。 但是，如果您使用的是函数模型，则在活页夹级别设置每个函数要容易得多，如我们在上面看到的那样。
对于生产部署，强烈建议通过配置显式指定应用程序 ID。 如果您要自动扩展应用程序，这一点尤其重要，在这种情况下，您需要确保使用相同的应用程序 ID 部署每个实例。
如果应用程序未提供应用程序 ID，则在这种情况下，活页夹将自动为您生成静态应用程序 ID。
这在开发方案中很方便，因为它避免了显式提供应用程序 ID 的需要。
以这种方式生成的应用程序 ID 将在应用程序重启时保持静态。
对于功能模型，生成的应用程序 ID 将是函数 bean 名称，后跟文字applicationID，例如：process-applicationID如果process如果函数 bean 名称。
设置应用程序 ID 的摘要
-
默认情况下，binder 将根据函数方法自动生成应用程序 ID。
-
如果您有单个处理器，则可以使用
spring.kafka.streams.applicationId,spring.application.name或spring.cloud.stream.kafka.streams.binder.applicationId. -
如果您有多个处理器，则可以使用属性 -
spring.cloud.stream.kafka.streams.binder.functions.<function-name>.applicationId.
使用功能样式覆盖活页夹生成的默认绑定名称
默认情况下，绑定器在使用函数式样式时使用上面讨论的策略来生成绑定名称，即 <function-bean-name>-<in>|<out>-[0..n]，例如 process-in-0、process-out-0 等。 如果要覆盖这些绑定名称，可以通过指定以下属性来实现。
spring.cloud.stream.function.bindings.<default binding name>.默认绑定名称是活页夹生成的原始绑定名称。
例如，假设您有这个函数。
@Bean
public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, Long>> process() {
...
}Binder 将生成带有名称的绑定，process-in-0,process-in-1和process-out-0.
现在，如果您想将它们完全更改为其他内容，也许是更多特定于域的绑定名称，那么您可以按如下方式进行作。
spring.cloud.stream.function.bindings.process-in-0=users
spring.cloud.stream.function.bindings.process-in-0=regions
和
spring.cloud.stream.function.bindings.process-out-0=clicks
之后，必须在这些新绑定名称上设置所有绑定级别属性。
请记住，使用上述函数式编程模型，在大多数情况下，遵循默认绑定名称是有意义的。 您可能仍希望执行此重写的唯一原因是，当您具有大量配置属性并且想要将绑定映射到更域友好的内容时。
设置引导服务器配置
运行 Kafka Streams 应用程序时，必须提供 Kafka 代理服务器信息。
如果您不提供此信息，则 Binder 希望您以默认值运行代理localhost:9092.
如果不是这种情况，那么您需要覆盖它。有几种方法可以做到这一点。
-
使用 boot 属性 -
spring.kafka.bootstrapServers -
活页夹级属性 -
spring.cloud.stream.kafka.streams.binder.brokers
当涉及到 binder 级别属性时，是否使用通过常规 Kafka binder 提供的 broker 属性并不重要 -spring.cloud.stream.kafka.binder.brokers.
Kafka Streams 绑定器将首先检查是否设置了 Kafka Streams 绑定器特定的代理属性 （spring.cloud.stream.kafka.streams.binder.brokers），如果未找到，则查找spring.cloud.stream.kafka.binder.brokers.