Spring @StreamListener进程(KStream<?，?>流)分区 [英] Spring @StreamListener process(KStream<?,?> stream) Partition

问题描述

我的流处理器中有一个包含多个分区的主题，我只是想从一个分区中进行流式处理，因此无法弄清楚如何配置它

I have a topic with multiple partitions in my stream processor i just wanted to stream that from one partition, and could nto figure out how to configure this

spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=s-processor
spring.cloud.stream.bindings.input.destination=uinput
spring.cloud.stream.bindings.input.group=r-processor
spring.cloud.stream.bindings.input.contentType=application/java-serialized-object
spring.cloud.stream.bindings.input.consumer.header-mode=raw
spring.cloud.stream.bindings.input.consumer.use-native-decoding=true

spring.cloud.stream.bindings.input.consumer.partitioned=true

@StreamListener(target = "input")
// @SendTo(value = { "uoutput" })
public void process(KStream<UUID, AModel> ustream) {

我只希望该处理器处理一个分区数据，其他分区将有其他处理器

I want only one partition data to be processed by this processor, there will be other processors for other partition(s)

到目前为止，我的发现与

So far my finding is something to do with https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/StreamsConfig.html#PARTITION_GROUPER_CLASS_CONFIG, but couldnot find how to set this property in spring application.properties

推荐答案

我认为分区分组程序是在单个处理器中将分区与任务分组.如果要确保处理器仅处理单个分区，则需要至少提供与主题分区相同数量的处理器实例.例如如果您的主题有4个分区，那么您需要有4个流应用程序实例，以确保每个实例仅处理一个分区.

I think the partition grouper is to group partition with tasks within a single processor. If you want to ensure that only a single partition is processed by a processor, then you need to provide at least the same number of processor instances as the topic partitions. For e.g. if your topic has 4 partitions, then you need to have 4 instances of the stream application to ensure that each instance is only processing a single partition.

这篇关于Spring @StreamListener进程(KStream<?，?>流)分区的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！