Node.js 14.x AWS Lambda中forEach循环内使用await的问题求助

问题原因 #

forEach 循环本身不支持异步操作的等待逻辑——当你在 forEach 里使用 await 时，循环会直接跳过当前等待，继续执行下一次迭代，不会等待异步任务完成，这就导致了Lambda运行时报错。

方案一：使用for...of 顺序处理 #

如果需要按顺序逐条处理SQS记录（比如依赖前一条处理结果，或者避免并发过高触发MQTT限流），用for...of是最直接的方式：

exports.handler = async (event) => {
  // 假设已在全局作用域初始化MQTT客户端client
  for (const record of event.Records) {
    try {
      // 解析SQS消息体
      const messageBody = JSON.parse(record.body);
      // 异步发布到MQTT主题
      await client.publish('your/mqtt/topic', JSON.stringify(messageBody));
      console.log(`消息 ${record.messageId} 发布成功`);
    } catch (error) {
      console.error(`处理消息 ${record.messageId} 失败:`, error);
      // 可选：抛出错误触发SQS重试，或根据业务逻辑处理失败场景
      throw error;
    }
  }
  return { statusCode: 200, body: '所有消息处理完成' };
};

方案二：使用Promise.all 并行处理 #

如果不需要严格顺序，想提升处理效率，可以用map生成所有发布任务的Promise，再用Promise.all等待全部完成：

exports.handler = async (event) => {
  // 假设已在全局作用域初始化MQTT客户端client
  const publishTasks = event.Records.map(async (record) => {
    try {
      const messageBody = JSON.parse(record.body);
      await client.publish('your/mqtt/topic', JSON.stringify(messageBody));
      return { messageId: record.messageId, status: 'success' };
    } catch (error) {
      console.error(`消息 ${record.messageId} 发布失败:`, error);
      return { messageId: record.messageId, status: 'failed', error: error.message };
    }
  });

  // 等待所有发布任务完成
  const results = await Promise.all(publishTasks);

  // 可选：统计失败消息
  const failedCount = results.filter(item => item.status === 'failed').length;
  if (failedCount > 0) {
    console.warn(`共 ${failedCount} 条消息发布失败`);
    // 可选：抛出错误触发重试，或返回失败状态
    // throw new Error('部分消息处理失败');
  }

  return { statusCode: 200, body: JSON.stringify(results) };
};

额外注意事项 #

• MQTT客户端初始化：建议在Lambda的全局作用域初始化客户端，避免每次函数调用都重新建立连接，提升性能。
• 幂等性：SQS会在Lambda执行失败时重试消息，确保你的MQTT发布逻辑是幂等的，避免重复发布导致业务异常。
• 并发限制：并行处理时要考虑MQTT broker的并发连接/消息限制，避免因并发过高被限流。

内容的提问来源于stack exchange，提问作者Gus Sabina