带有 GCP pubsub 的 Spring Cloud Stream 使用者的并发设置

Question

我的应用程序正在使用绑定到 GCP pubsub 的 Spring Cloud Stream 接收消息。我正在试验以下配置参数：

spring.cloud.gcp.pubsub.subscriber.executor-threads
spring.cloud.stream.bindings.<channelName>.consumer.concurrency

该应用程序有 3 个不同的通道，每个通道都定义了一个消费者组。应用程序的多个实例将在生产中运行（在 kubernetes 上）。

我正在尝试找到正确的设置进行配置，以确定在每个应用程序实例中可以并行处理多少消息。我一直在我的本地机器上尝试调整上述两个参数，但似乎只有 executor-threads 有任何影响。如果我将其设置为 5，并将一堆消息泵入系统，我会在消息处理日志中看到 5 个线程。如果我将它提高到 10，我会在那里看到 10 个线程。然而，并发参数似乎没有任何作用，无论它是设置为 1 还是 10 或其他什么。

如果有的话，这些参数之间的关系是什么？并发参数是否仅用于 Rabbit 或 Kafka 等其他绑定器？

谢谢。

Answer 1

spring.cloud.gcp.pubsub.subscriber.executor-threads
spring.cloud.gcp.pubsub.subscriber.parallel-pull-count

Spring Cloud GCP中的这两个配置选项setExecutorThreadCount与setParallelPullCountPub/Sub 的Java 客户端库中的相同。它们控制您的Pub/Sub订阅者客户端的行为。

parallel-pull-count确定您的订阅者客户端打开多少个流来接收消息，并executor-threads确定您的订阅者客户端使用多少个线程来处理消息回调（消息被确认以便 Pub/Sub 服务不会再次将它们发送给您）。例如，当您设置parallel-pull-count为 2 和executor-threads4 时，消息将通过 2 个流传入，并且 8 个线程将忙于处理它们。有关更多信息，请参阅并发控制。

spring.cloud.stream.bindings.<channelName>.consumer.concurrency

您是对的，此配置选项仅适用于某些活页夹。为了在您的消费者组中设置并发性，您的生产者必须进行分区。但是，Spring Cloud Stream binder to Pub/Sub 目前不支持分区，因此您将无法设置并发。

我正在尝试找到正确的设置进行配置，以确定在每个应用程序实例中可以并行处理多少消息。

为了实现最大吞吐量，我将使用上面的前两个配置选项，同时还要记住，处理时间超过其确认截止日期的消息将过期，并且 Pub/Sub 服务将重新传递它们。