JS 中的 Gcloud pubsub worker

Question

我正在使用 setInterval 在 JS 中开发一个 worker，每 10 秒检查一次是否有新消息。我想知道以下代码的方法是否正确。你认为我会有一些性能问题吗？我在 Kubernetes 中将它作为单个 pod 运行。

const messageHandler = message => {
  sendRequest(message, (message, error, response, body) => {
    if (!error && response.statusCode == 200) {
      message.ack();
    }
  });
};

subscription.on(`message`, messageHandler);

const timeout = 10;
setInterval(() => {
  subscription.removeListener('message', messageHandler);
  subscription.on(`message`, messageHandler);
}, timeout * 1000);

Answer 1

我可能在这里遗漏了一些东西，但如果您希望您的订阅者“永远”启动并收听，那么我完全不明白您为什么需要删除您的听众。

那么为什么不这样呢：

const messageHandler = async message => {
    try {
        await doSomethingWith(message);
        message.ack();
    } catch (err) {
        console.error(`Error handling message: ${message.id}: ${err.message}`);
        message.nack();
    }
};


// Listen for new messages
subscription.on(`message`, messageHandler);

Answer 2

不是每 10 秒删除和注册你的 messageHandler 侦听器，以下是两种更好的方法：

1. 对传入消息有一个长时间运行的消息侦听器，直到自创建侦听器以来达到“x”时间。
2. 为传入消息设置一个长时间运行的消息侦听器，并且仅在自收到最后一条消息后经过“x”时间后才关闭订阅者。

在消息延迟的极少数情况下，订阅者可能会在收到所有已发布的消息之前关闭。但是，如果您将超时设置为一个很大的数字（以分钟为单位），这不太可能发生。

有关创建订阅者客户端的示例，请参阅client library。您可以将超时修改为更大的数字。客户端库使用StreamingPull，它维护一个开放的双向流并在消息可用时接收消息，以实现最大吞吐量和低延迟。