使用 pub sub 调用函数时如何避免内存泄漏？答案

【问题标题】：How to avoid memory leak when using pub sub to call function?使用 pub sub 调用函数时如何避免内存泄漏？
【发布时间】：2022-06-22 18:38:55
【问题描述】：

我在使用 pubsub 触发功能时遇到了性能问题。

//this will call on index.ts
export function downloadService() {
  // References an existing subscription
  const subscription = pubsub.subscription("DOWNLOAD-sub");

  // Create an event handler to handle messages
  // let messageCount = 0;
  const messageHandler = async (message : any) => {
    console.log(`Received message ${message.id}:`);
    console.log(`\tData: ${message.data}`);
    console.log(`\tAttributes: ${message.attributes.type}`);

    // "Ack" (acknowledge receipt of) the message
    message.ack();
    await exportExcel(message);//my function
    // messageCount += 1;
  };


  // Listen for new messages until timeout is hit
  subscription.on("message", messageHandler);
} 

async function exportExcel(message : any) {
 //get data from database
 const movies = await Sales.findAll({
    attributes: [
      "SALES_STORE",
      "SALES_CTRNO",
      "SALES_TRANSNO",
      "SALES_STATUS",
    ],
    raw: true,
  });
  ... processing to excel// 800k rows
  ... bucket.upload to gcs

}

如果我只触发一条 pubsub 消息，上面的函数就可以正常工作。但是，如果我在短时间内触发许多 pubsub 消息，该函数会遇到内存泄漏问题或数据库连接超时问题。

我发现的问题是，第一次处理还没有完成，但是来自pubsub的其他请求将直接再次调用函数并同时处理。

我不知道如何解决这个问题，但我在想实现队列工作者或谷歌云任务将解决问题？

【问题讨论】：

不确定，但为什么要等待 excel 调用？我猜这就是它吃掉你所有记忆的地方。只需调用它并继续（无需等待）或将它们放入队列中以便稍后处理。添加此 // await exportExcel(message); 或者如果您使用的是 Deno，您可以使用并行映射一次处理 N 个。
@chovy 嗨，我试过删除“等待”，但还是一样。实际上，当这个函数同时调用上面的 3 时，处理到 excel 时会发生内存问题。好吧，让我试试看。
是的...这是一个繁重的过程。您需要将它们排队。如果您使用的是节点，那里有一些队列库。我忘记了哪些。但是你可以说“处理这个数组，直到它一次做 X 为空”

标签： node.js express publish-subscribe google-cloud-pubsub google-cloud-tasks

【解决方案1】：

正如@chovy 在 cmets 中所提到的，需要将excelExport 函数调用排队，因为函数的执行跟不上调用速率。可用于对函数调用进行排队的模块之一是async。请注意，async 模块不受 Google 官方支持。

作为替代方案，您可以在订阅者端使用flow control 功能。数据管道通常会在已发布的流量中收到零星的峰值，这可能会使订阅者不堪重负以迎头赶上。对订阅的高发布吞吐量的通常响应是动态自动缩放订阅者资源以消耗更多消息。但是，这可能会产生不必要的成本——例如，您可能需要使用更多的虚拟机——这可能会导致额外的容量规划。订阅者端的流控制功能可以通过允许订阅者调节接收消息的速率来帮助控制管道上这些任务的不健康行为。请参阅此blog 了解有关流量控制功能的更多信息。

【讨论】：