谷歌云 pubsub node.js 客户端与谷歌云功能不兼容

Question

架构：

我们有一个使用 2 个发布订阅主题/订阅对的架构：

• 主题T1 由cronjob 定期触发（例如每5 分钟一次）。订阅S1 是我们云功能的触发器。
• 主题T2 用作我们的一项服务发布的后台作业的队列。订阅S2 由云函数在每次执行时读取，以服务排队的后台作业。

这使我们能够控制后台作业的服务频率，而与它们何时添加到队列无关。

云函数（由S1 触发）通过pulling 读取来自S2 的消息。它决定哪些后台作业已准备就绪，并在成功为作业提供服务后，它会确认相关的消息。未准备好或失败的作业不会被确认以供以后提供服务。

问题：

我们在使用来自 google 的官方 node.js pubusb client 时遇到问题：

1. 有时 ACK 的消息会重新出现（似乎是无限的）。我们验证了消息在 ACK 截止日期之前得到确认，并通过调查我们的日志确定我们正在致电 ack()。
2. 有时在第一次执行后（重新部署函数后），后续执行永远不会收到新消息。我们可以通过验证堆栈驱动程序中未确认的消息计数或通过重新部署函数并查看消息得到服务来验证消息是否在订阅S2 中排队。

我们认为这是 google 的 node.js pubsub 客户端的问题。云功能文档明确指出not start background activities。但是，查看 node.js pubsub 客户端源代码，它显然使用超时在后台服务确认。

google的node.js pubsub客户端不兼容google cloud的功能吗？谷歌recommends accessing the service API's only when a client library does not exist or does not meet other needs。在云功能中运行客户端是否“其他需求”，需要我们使用服务 API 编写自己的客户端？

尝试的解决方法：

作为“解决方法”，我们尝试延迟执行 cloudfunction 的结束，以允许 node.js pubsub 客户端中的任何“后台”进程完成，但这并没有始终消除我们的问题。 pubsub 客户端似乎对云功能不友好，并且无法从云功能执行之间的停止中恢复。

2018 年 2 月 22 日更新

我写了an article on our blog，详细描述了为什么我们以这种方式使用 PubSub，以及我们如何解决 node.js pubsub 客户端与云功能不兼容的事实。

Answer 1

我遇到了同样的问题，我想更好地控制.ack()。查看 google 的 nodejs 库，可以选择重构 ack() 以返回承诺，以便函数可以等待 ack() 完成。

Subscriber.prototype.ack_ = function(message) {
  var breakLease = this.breakLease_.bind(this, message);

  this.histogram.add(Date.now() - message.received);

  if (this.writeToStreams_ && this.isConnected_()) {
    this.acknowledge_(message.ackId, message.connectionId).then(breakLease);
    return;
  }

  this.inventory_.ack.push(message.ackId);
  this.setFlushTimeout_().then(breakLease);
};

Answer 2

来自 node.js pubsub 客户端 confirmed 的开发人员表示不支持使用客户端从云函数中提取消息。

替代方法是使用service APIs。但是，当尝试从订阅中提取 所有 消息时，REST API 具有 their own caveats。

Answer 3

你是如何触发你的函数的？

根据docs，如果您的函数正在使用 pubsub 消息，那么您应该使用 pubsub 触发器。使用 pubsub 触发器时，不需要该库。只需在函数结束时调用callback()，pubsub 消息就会被正确确认。

对于您打算做的事情，我认为您当前的架构不是正确的选择。

我会使用cron task 将您的第一步移至 Google App Engine，并让此任务只需将消息从 T2 移动到 T1，让函数具有触发器 S2 并处理消息。

因此，您的工作将发布在 T2，并且您将拥有一个 GAE 应用程序，该应用程序具有由 cron 任务触发的拉订阅 S2，并且此应用程序会将消息重新发布到 T1。然后，您的函数将由订阅 S1 到主题 T1 触发，并在消息中运行作业，避免导入 pubsub 库的额外处理，并按预期使用产品。

此外，我不确定您最初是如何将您的工作发布到该主题的，但 Task Queues 是用于速率限制任务的一个很好的 GAE（和 product-agnostic in Alpha）选项。

仅用于此目的的 GAE 应用（设置 1 个最大实例）将位于 always free limit 内，因此成本不会显着增加。