如何对 Google Cloud Pub/Sub 队列进行速率限制答案

【问题标题】：How To Rate-Limit Google Cloud Pub/Sub Queue如何对 Google Cloud Pub/Sub 队列进行速率限制
【发布时间】：2021-09-30 00:55:45
【问题描述】：

我正在使用 Google 的 Pub/Sub 队列来处理服务之间的消息。一些订阅者连接到速率限制 API。

例如，我将街道地址推送到发布/订阅主题。我有一个订阅（通过推送）该主题的云功能，并调用外部限速地理编码服务。理想情况下，我的街道地址可以立即推送到主题上，并且主题会保留这些消息 - 以限速方式呼叫订阅者。

有没有配置这样的延迟，或消息分发速率限制？增加 Ack 窗口并没有真正的帮助：我已经构建了这个系统来防止长时间运行的功能。

【问题讨论】：

我遇到了 PubSub 和 Cloud Functions 的速率限制问题。解决方案是创建要在数据存储中完成的作业记录。在从 Datastore 中提取 n 个对象的调度程序上创建一个云函数，然后执行一个云函数。您正在有效地创建一个排队系统。在流程中的每一步更新 DS 对象是一件很痛苦的事情，但它也可以轻松地可视化管道。
一年，有更新吗？
@slideshowp2 google 似乎对支持这个用例不感兴趣，没有更新
您可以尝试定期从 Pub/Sub 拉取，使用 Cloud Scheduler 您可以设置一个 cron 来触发 Cloud Function 从 Pub/Sub 主题拉取有限数量的消息并调用您的速率限制API。
三年后，上述方法似乎是正确的方法（通过 cron 作业定期手动从队列中拉出）

标签： node.js google-cloud-functions google-cloud-pubsub

【解决方案1】：

解决问题的方法是使用：async.queue

你有一个并发属性，你可以管理速率限制。

// create a queue object with concurrency 2
var q = async.queue(function(task, callback) {
    console.log('hello ' + task.name);
    callback();
}, 2);

// assign a callback
q.drain = function() {
    console.log('all items have been processed');
};

// add some items to the queue
q.push({name: 'foo'}, function(err) {
    console.log('finished processing foo');
});

// quoted from async documentation

【讨论】：

我的目标是将任务存储在 pub/sub 队列中（想想 SNS/SQS），这样我就不必拥有一个持续运行的服务器，而是可以使用短命的 lambdas。此解决方案需要一个长时间运行的服务器。
此外，在GKE 集群上部署应用程序怎么样？这意味着应用程序有多个实例。

【解决方案2】：

因为到目前为止还没有描述解决方法的答案，所以我现在将通过说明目前没有办法做到这一点来回答这个问题。有一些变通方法（请参阅问题上的 cmets，解释如何使用 Cloud Scheduler 创建排队系统），但无法仅在拉取订阅上设置一个设置，从而在其与其主题之间创建速率限制。

我为此打开了一个功能请求。如果您喜欢此功能，请就跟踪的问题发表意见。

https://issuetracker.google.com/issues/197906331

【讨论】：

4 年后我认为这不会发生。似乎 pub/sub 是一个队列，仅此而已 - 该队列的消费超出了此抽象的范围。事实上，AWS SQS 与要求在消费者端进行节流类似。
好吧，我确实看到 GCP 最近添加了很多功能。我们得到了死信主题，在主题级别强制执行 Protobuf 模式。这是好东西。也许我们会在 GCP 管理的消费者端受到限制。
更新：功能请求已进入“已接受”状态。