这个版本还在开发中,还不算稳定。对于最新的稳定版本,请使用 spring-cloud-stream 4.3.0! |
出站分区支持
Kafka Streams 处理器通常将处理后的输出发送到出站 Kafka 主题中。如果出站主题是分区的,并且处理器需要将传出数据发送到特定分区中,那么应用程序需要提供类型为StreamPartitioner
. 有关更多详细信息,请参阅 StreamPartitioner。让我们看一些示例。
这是我们已经多次看到的同一个处理器,
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>> process() {
...
}
以下是输出绑定目标:
spring.cloud.stream.bindings.process-out-0.destination: outputTopic
如果主题outputTopic
有 4 个分区,如果您不提供分区策略,Kafka Streams 将使用默认的分区策略,这可能不是您想要的结果,具体取决于特定用例。假设您想发送任何匹配的密钥spring
对 0 进行分区,cloud
到分区 1,stream
分区 2,其他所有内容都分区 3。这是您需要在应用程序中执行的作。
@Bean
public StreamPartitioner<String, WordCount> streamPartitioner() {
return (t, k, v, n) -> {
if (k.equals("spring")) {
return 0;
}
else if (k.equals("cloud")) {
return 1;
}
else if (k.equals("stream")) {
return 2;
}
else {
return 3;
}
};
}
这是一个基本的实现,但是,您可以访问记录的键/值、主题名称和分区总数。因此,如果需要,您可以实现复杂的分区策略。
您还需要提供此 Bean 名称以及应用程序配置。
spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitionerBeanName: streamPartitioner
应用程序中的每个输出主题都需要像这样单独配置。