春云流 Kafka 消费者并发

Question

使用 spring-cloud-stream 的 kafka binder，如何配置并发消息消费者（在单个消费者 jvm 中）？如果我理解正确，使用 kafka 时并发消息消耗需要分区，但 s-c-s 文档表明要使用分区，您需要通过分区键表达式或分区键提取器类在生产者中指定分区选择。Kafka 文档提到了循环分区。

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

我

似乎得到了我想要的行为（至少在某种程度上）。我可以看到有时有 3 个消费者线程处于活动状态，尽管除了轮循机制之外似乎确实有一些分区在起作用，因为某些消息似乎等待繁忙的消费者线程并在该线程完成后被消耗。我认为这是因为消息被发送到同一个分区。

当我不指定分区键表达式或分区键提取器类时，是否有一些默认的键提取和分区策略在生产者上使用？这是使用 kafka 设置 s-c-s 使用者的合适方法，您希望多个线程使用消息以提高使用者吞吐量？

Answer 1

由于您的生产者未分区（没有设置partitionKeyExpression），因此生产者端将对 3 个分区进行轮询（如果这不是观察到的行为，请在 Git Hub 中打开票证）。如果配置了partitionKeyExpression则创建器将根据配置的逻辑有效地对数据进行分区。

在消费者方面，我们确保线程/分区亲和性，因为这是一个广受推崇的 Kafka 约定 - 我们确保给定分区上的消息按顺序处理 - 这可能解释了您观察到的行为。如果将消息 A、B、C、D 发送到分区 0,1,2,0 - D 必须等到 A 被处理，即使还有另外两个线程可用。

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

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

一般来说，partitioned是指客户端接收分区数据的能力（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）