spring-cloud-stream - 使用 GCP pubsub 的 Spring Cloud Stream 消费者的并发设置

Question

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

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

该应用程序有 3 个不同的 channel ，每个 channel 都定义了一个使用者组。应用程序的多个实例将在生产环境中运行(在 kubernetes 上)。

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

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

谢谢。

Answer 1

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

Spring Cloud GCP 中的这两个配置选项与 setExecutorThreadCount 相同和 setParallelPullCount在 Pub/Sub 的 Java client library .他们控制着行为
您的 Pub/Sub订户客户端。
parallel-pull-count确定您的订阅者客户端打开多少个流来接收消息和 executor-threads确定您的订阅者客户端使用多少线程来处理消息回调(其中消息被确认，因此 Pub/Sub 服务不会再次将它们发送给您)。例如，当您设置 parallel-pull-count到 2 和 executor-threads到 4，消息将通过 2 个流进入，8 个线程将忙于处理它们。见 Concurrency Control了解更多信息。

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

您是对的，此配置选项仅适用于某些活页夹。为了在你的消费者组中设置并发，你的生产者必须被分区。但是，Pub/Sub 的 Spring Cloud Stream 绑定(bind)器当前执行 not support partitioning ，因此您将无法设置并发。

I'm trying to find the right settings to configure to determine how many messages can be processed in parallel in each app instance.

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