在 Spring Cloud GCP pubsub 中创建特定于消息通道的线程

Question

我有一个运行 GCP PubSub 消息传递的 Spring Cloud 应用程序。我有 2 个消息入站频道订阅了 2 个不同的订阅者。我在应用程序的负载/压力测试期间面临的问题是，特定的线程数设置如下：

spring.cloud.gcp.pubsub.subscriber.executor-threads: 350
spring.cloud.gcp.pubsub.subscriber.parallel-pull-count: 2
spring.cloud.gcp.pubsub.subscriber.max-acknowledgement-threads: 700

当通道 1 的消息拉取的进程繁忙时，我没有足够的线程让通道 2 拉取消息。解决方案是限制/配置每个通道的线程数。我发现很难弄清楚这一点。请帮帮我！以下是我所指的频道：

@Bean
@ServiceActivator(inputChannel = "pubsubInputChannel1")
public MessageHandler extractionMessageReceiver() {
    return message -> {
        // do something
    };
}

@Bean
@ServiceActivator(inputChannel = "pubsubInputChannel2")
public MessageHandler extractionMessageReceiver() {
    return message -> {
        // do something
    };
}

注意，订阅者线程会一直忙到某个特定进程被消息拉取结束。

Answer 1

我遇到了以下问题：当有很多消息并且它们排队时，pubsub 的执行器健康检查停止工作。我的假设是所有执行程序线程都忙于处理消息，并且检查运行到最后期​​限超出异常。 以下流控制属性帮助我解决了这个问题：

Config	Description
spring.cloud.gcp.pubsub.[subscriber,publisher.batching].flow-control.max-outstanding-element-count	Maximum number of outstanding elements to keep in memory before enforcing flow control
spring.cloud.gcp.pubsub.[subscriber,publisher.batching].flow-control.max-outstanding-request-bytes	Maximum number of outstanding bytes to keep in memory before enforcing flow control.

来自https://docs.spring.io/spring-cloud-gcp/docs/1.1.0.M1/reference/html/_spring_cloud_gcp_for_pub_sub.html

当我将最大未完成元素数设置为 100 时，一切正常。

我认为未完成的消息是通过流来提取的。通过上面的属性，我们可以控制不是所有的消息都被一次处理。相反，我们将它们分成例如每条 100 条消息。也许它也会在频道之间切换。引用https://medium.com/google-cloud/things-i-wish-i-knew-about-google-cloud-pub-sub-part-2-b037f1f08318:

请注意，流式拉取只能确保尽最大努力进行流量控制。假设您注意到您的应用程序在任何一个时间段内只能处理 100 条消息，因此您将最大未完成消息设置为 100。客户端将在收到 100 条消息后暂停，这在大多数情况下都有效。但是，如果您随后在单个发布批次中发布 500 条消息，客户端将一次收到所有 500 条消息，但一次只能处理 100 条，这可能导致过期消息积压不断增加。这是因为流式拉取无法从单个发布批次中拆分消息。为避免这种情况，请在发布时增加订阅者数量或减少批量大小以匹配订阅者消息处理能力。

这些参数能解决你的问题吗？