Spring@StreamListener过程（KStream流）分区

Question

我的流处理器中有一个关于多个分区的主题，我只是想从一个分区进行流处理，但是不知道如何配置

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) {

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

到目前为止，我的发现与https://Kafka . Apache . org/20/javadoc/org/Apache/Kafka/streams/streams CONFIG . html # PARTITION _ GROUPER _ CLASS _ CONFIG有关，但无法找到如何在spring application.properties中设置该属性

Answer 1

Kafka Streams不允许读取单个分区。如果您订阅了一个主题，所有分区都会被使用并分布在可用的实例上。因此，您无法提前知道哪个分区被分配给了哪个实例，并且所有实例都执行相同的代码。

但链接到处理器的每个分区都有不同类型的数据，因此需要不同的处理器应用程序

对于这种情况，处理器(或转换器)必须能够处理所有分区的数据。Kafka Streams通过< code>ProcessorContext对象公开分区号，该对象通过< code>init()方法传递给处理器:https://Kafka . Apache . org/20/javadoc/org/Apache/Kafka/Streams/kstream/transformer . html # init-org . Apache . Kafka . Streams . processor context-

因此，您需要在变压器内“分支”以根据分区应用不同的处理逻辑：

ustream.transform(() -> new MyTransformer());


class MyTransformer implement Transformer {
  // other methods omitted

  R transform(K key, V value) {
    switch(context.partition()) { // get context from `init()`
      case 0:
        // your processing logic
        break;
      case 1:
        // your processing logic
        break;

      // ...
  }
}

Answer 2

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