spring-cloud-stream kafka消费者并发

Question

使用Spring-Cloud-Stream的kafka绑定器，如何配置并发消息消费者（在单个消费者jvm中）？如果我没有理解错的话，在使用kafka时并发使用消息需要分区，但是s-c-s文档指出，要使用分区，您需要通过partitionKeyExpression或PartitionKeyExtractorClass在生成器中指定分区选择。Kafka博士提到html" target="_blank">循环分区。

s-c-s文档根本没有提到spring.cloud.stream.bindings.*.concurrency，尽管这在我上面描述的用例中似乎很重要。带有生产者配置

spring:
  cloud:
    stream:
      bindings:
        customer-save:
          destination: customer-save
          group: customer-save
          content-type: application/json
          partitionCount: 3

和使用者配置

spring:
  cloud:
    stream:
      bindings:
        customer-save: 
          destination: customer-save
          group: customer-save
          content-type: application/x-java-object;type=foo.Customer
          partitioned: true
          concurrency: 3

当我没有指定partitionKeyExpression或PartitionKeyExtractorClass时，是否有一些默认的键提取和分区策略在生产者上使用？这是一种使用kafka设置s-c-s消费者的适当方法吗？在kafka中，您需要多个线程来消费消息，以提高消费者吞吐量？

Answer 1

由于您的生产者没有分区（没有partitionkeyexpression),生产者端将在3个分区上循环（如果这不是观察到的行为,请在Git Hub中打开一个票证）。如果您配置了PartitionKeyExpression，那么生成器将根据配置的逻辑有效地对数据进行分区。

在消费者端，我们确保线程/分区的亲和性，因为这是一个广受尊重的Kafka约定--我们确保给定分区上的消息是按顺序处理的--这可能是您所观察到的行为的原因。如果将消息A、B、C、D发送到分区0、1、2、0-D将不得不等到A被处理完，即使有另外两个线程可用。

增加吞吐量的一个选择是过分区（这是Kafka中相当典型的策略）。这将进一步分散消息，并增加消息发送到不同线程的机会。

如果您不关心排序，增加吞吐量的另一个选择是在下游异步处理消息：例如，通过将输入通道桥接到ExecutorChannel。

一般来说，partitied是指客户端接收分区数据的能力（Kafka客户端总是分区的，但此设置也适用于Rabbit和/或Redis）。它与属性instanceIndex和instanceCount结合使用，以确保主题的分区在多个应用程序实例之间正确划分（另请参见http://docs.spring.io/spring-cloud-stream/docs/1.0.0.m4/reference/htmlsingle/index.html#_instance_index_and_instance_count)