如何确保Readable流中所有异步数据库插入完成后再resolve Promise？

你的问题核心在于：Node.js流的end事件仅表示文件内容已读取完毕，但不会等待data事件中触发的异步数据库操作完成，导致Promise提前resolve，而数据库任务还在后台执行。

下面提供两种靠谱的解决思路：

方案一：使用异步迭代器（for await...of） #

这是最简洁现代的写法，直接遍历流的异步迭代器，确保每一行的数据库操作完成后再处理下一行：

async function insertToDb(data) {
  await repository.save(new Entity(data.id, data.name, data.address)); // 注意替换为读取到的data字段，不要写死值
}

async function processFile(buffer) {
  const stream = Readable.from(buffer);
  // 顺序处理每一行，等待当前插入完成再处理下一行
  for await (const data of stream) {
    await insertToDb(data);
  }
  return 1; // 所有操作完成后返回
}

如果需要提高效率，支持并发插入（比如同时处理5行），可以加个并发控制：

async function processFile(buffer, concurrency = 5) {
  const stream = Readable.from(buffer);
  const pendingPromises = [];

  for await (const data of stream) {
    const promise = insertToDb(data);
    pendingPromises.push(promise);

    // 达到并发上限时，等待最早完成的一个任务
    if (pendingPromises.length >= concurrency) {
      const finished = await Promise.race(pendingPromises);
      // 移除已完成的任务
      pendingPromises.splice(pendingPromises.indexOf(finished), 1);
    }
  }

  // 等待所有剩余的插入任务完成
  await Promise.all(pendingPromises);
  return 1;
}

方案二：用计数器跟踪异步任务状态 #

如果不想改动原有事件监听的写法，可以通过计数器记录未完成的数据库任务，只有当流结束且所有任务都完成时才resolve：

async function insertToDb(data) {
  await repository.save(new Entity(data.id, data.name, data.address));
}

return new Promise((resolve) => {
  let pendingTasks = 0;
  let hasStreamEnded = false;

  // 检查是否可以resolve的逻辑
  const tryResolve = () => {
    if (hasStreamEnded && pendingTasks === 0) {
      resolve(1);
    }
  };

  Readable.from(buffer)
    .on('data', async (data) => {
      pendingTasks++;
      try {
        await insertToDb(data);
      } finally {
        // 不管成功失败，都减少计数器并检查是否可以resolve
        pendingTasks--;
        tryResolve();
      }
    })
    .on('end', () => {
      hasStreamEnded = true;
      tryResolve();
    });
});

注意点 #

• 你的insertToDb里创建Entity时用了写死的"id"、"name"，实际应该替换成读取到的data中的对应字段，不然所有插入的数据都是一样的。
• 如果数据库操作可能失败，建议在insertToDb里添加错误处理，避免单个失败导致整个流程中断（根据业务需求选择重试或跳过）。

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