此版本仍在开发中,尚不被认为是稳定的。对于最新的稳定版本,请使用 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
.