Spring @StreamListener process(KStream<?,?>stream) 分区 [英] 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)

到目前为止，我的发现与 https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/StreamsConfig.html#PARTITION_GROUPER_CLASS_CONFIG，但是在spring application.properties中找不到如何设置这个属性

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 process(KStream<?,?>stream) 分区的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！