GCP Nodejs8 云功能 - 同步 PubSub 发布

Posted 2023-02-16

【中文标题】GCP Nodejs8 云功能 - 同步 PubSub 发布

【英文标题】：GCP Nodejs8 Cloud Function - Synchronous PubSub publish

【发布时间】：2020-02-06 21:30:05

【问题描述】：

我正在努力使用 javascript/Nodejs8 Google Cloud Function 将有效负载发布到 Google PubSub。

所以我有一个由 HTTP 请求触发的云函数，然后请求正文被发布到一个 pubsub 主题（配置为拉模式）。

这是我的代码：

const PubSub = require('@google-cloud/pubsub');
const pubsub = new PubSub();
const topic = pubsub.topic('my-fancy-topic');

function formatPubSubMessage(reqObj)
    // the body is pure text
    return Buffer.from(reqObj.body);
;

exports.entryPoint = function validate(req, res) 

topic.publish(formatPubSubMessage(req)).then((messageId) => 
            console.log("sent pubsub message with id :: " + messageId)
        );

res.status(200).json("res":"OK");
;

我的问题是云函数在发布 pubsub 消息之前完成执行（在日志中，日志“函数执行需要 X 毫秒，以状态代码完成：200”在我的 pubsub 日志之前大约 30 或 40 秒出现. 我也有几次“忽略完成函数中的异常”的日志，但我没有得到我的 pubsub 日志）

我不是 javascript 或 nodejs 专家，我也不掌握 javascript 承诺，但我想知道是否可以使发布同步。我也在想我可能在这里做错了什么！

提前感谢您的帮助。

【参考方案1】：

在您的逻辑中，您的回调/事件处理函数在 HTTP 消息到达时被调用。然后执行一个 publish() 函数。执行发布是一个异步活动。这意味着发布需要一些时间才能完成，并且由于 JavaScript（本质上）不想阻塞，它会立即返回一个 promise，然后您可以在异步工作完成时收到通知。在执行 publish() 之后，您的逻辑会立即执行 res.status(....) ，它会向 HTTP 请求发送响应，这确实是来自 HTTP 客户端的流请求的结束。异步发布仍在进行中，当它本身完成时，发布的回调就会发生并记录响应。

不幸的是，这不是谷歌在此处记录的好习惯...

https://cloud.google.com/functions/docs/bestpractices/tips#do_not_start_background_activities

在最后一个故事中，您调用的函数validate 仍将在发布完成之前结束。如果您想在 publish() 执行时阻塞（有效地使其同步），您可以使用 JavaScript await 关键字。大致如下：

try 
   let messageId = await topic.publish(....);
   console.log(...);
catch(e) 
 ...

您还需要将函数标记为async。例如：

exports.entryPoint = async function validate(req, res) 
   ...

见：https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/async_function

您也可以简单地从函数中返回一个 Promise，并且在整个 Promise 解决之前，回调函数不会被视为已解决。

底线是深入研究 Promise。

【讨论】：

感谢您的回答@Kolban。您能否详细说明如何将函数标记为异步以实现您的解决方案 plz？再次感谢您。

更新答案以包含异步信息

感谢您的贡献。我可以弄清楚，现在它同步运行。 @Doug 的回答有效，但我接受你的回答，因为它更“灵活”，而且它是我得到的第一个答案。再次感谢

很高兴它可以提供帮助。我对这些观点并不大惊小怪。随意将@Doug 的答案识别为正确答案。如果他对你的理解有所帮助，你可以考虑投票给他的答案。

【参考方案2】：

现在，此代码在发布完成之前发送响应。发送响应后，函数终止，正在进行的异步工作可能无法完成。

您应该做的是仅在发布完成后才发送响应，这意味着将该行代码放入 then 回调中。

exports.entryPoint = function validate(req, res) 

    topic.publish(formatPubSubMessage(req)).then((messageId) => 
        console.log("sent pubsub message with id :: " + messageId)
        res.status(200).json("res":"OK");
    );

;

我建议花一些时间了解 Promise 的工作原理，因为这对于构建正确工作的函数至关重要。

【讨论】：

感谢您的回答道格。我一定会花时间研究 javascript 的 promise。在接受任一答案之前，我会尝试您的解决方案和 Kolban 的解决方案。