使用async parallelLimit仅执行一次就停止，如何处理到数组遍历完成？

• 核心问题是async/parallelLimit的任务函数默认适配回调风格，你当前传入的是返回Promise的异步函数，没有触发任务完成的回调通知，导致并行调度器认为任务一直未完成，无法调度后续任务执行。
• 次要问题：4000的并发阈值过高，远超大多数云服务（比如你用到的Pub/Sub）的默认QPS限制，很容易触发限流报错导致执行中断。

提供两种可选修复方式，可根据实际场景选择：

方案1：适配async库规范修改代码 #

使用async库自带的asyncify包装Promise风格的任务函数，让调度器可以正确识别任务完成状态：

const parallelLimit = require("async/parallelLimit");
const { asyncify } = require("async");

async function publishMessage(data) {
  try {
    const messageId = await pubSubClient
      .topic(topicName)
      .publishJSON(data);
    console.log(messageId);
    return messageId;
  } catch (err) {
    console.log(err);
    throw err; // 错误抛出后可被parallelLimit统一捕获处理
  }
}
messages = messages.flat();
// 用asyncify包装返回Promise的函数，适配async库的回调调度逻辑
let tasks = messages.map((row) => asyncify(() => publishMessage(row)));

// 并发数建议调整到Pub/Sub允许的范围内，比如100-500，避免触发限流
await parallelLimit(tasks, 200);

方案2：原生JS实现并发限制，无需依赖async库 #

如果不想额外适配async库的规则，可以自己实现简单的并发限制逻辑，直接兼容Promise风格的任务：

// 原生实现并发限制工具函数
async function parallelLimit(tasks, limit) {
  const result = [];
  const executing = [];
  for (const task of tasks) {
    const p = Promise.resolve().then(() => task());
    result.push(p);
    if (limit <= tasks.length) {
      const e = p.then(() => executing.splice(executing.indexOf(e), 1));
      executing.push(e);
      if (executing.length >= limit) {
        await Promise.race(executing);
      }
    }
  }
  return Promise.all(result);
}

// 原有业务逻辑无需额外适配，直接调用即可
async function publishMessage(data) {
  try {
    const messageId = await pubSubClient
      .topic(topicName)
      .publishJSON(data);
    console.log(messageId);
    return messageId;
  } catch (err) {
    console.log(err);
  }
}

messages = messages.flat();
let tasks = messages.map((row) => () => publishMessage(row));
await parallelLimit(tasks, 200);

• 建议结合业务场景和下游服务的限流规则调整并发数，避免因请求被拒绝导致任务失败
• 如果需要保留失败任务的重试逻辑，可以在publishMessage内部增加重试机制，避免单个消息发送失败影响整体执行

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