Promise.all并行数据库请求的阻塞问题及顺序保持方案

我本来期望Promise.all能并行处理每个传入的数据项，但实际运行时，所有“MAPPING DATA”日志打印完成后，到“CHECKED DB”日志开始打印前有明显停顿。看起来const snap = await global.db.collection("Prompts").doc(id).get();并没有被并行执行，而是等所有映射完成后才开始执行数据库查询。

另外我有个疑问：同样用了await的global.db.collection("collection").doc(id).set(data);似乎不会阻塞进程，console.log("WRITE FOR: "+id)能持续输出，这是为什么？

最后我想知道：如何在并行执行这些请求的同时，保持每个数据项的执行顺序（先检查数据库，再按需写入）？

注：当前测试涉及10K条文档写入。

现有代码 #

async function parallelIndividualWrites(datas) {

  try {
    let mapped = 1;
    let dbcheck = 1;
    let counter = 1;
    await Promise.all(datas.map(async (data) =>  {
         // ...
      console.log("MAPPING DATA: "+(mapped++)+"/"+datas.length);
      const snap = await global.db.collection("Prompts").doc(id).get();
      console.log("CHECKED DB: "+(dbcheck++)+"/"+datas.length);
      if (snap.data()) {
          console.log("NO WRITE FOR: "+id);
          console.log("COUNT: "+counter++);
      }
      else {
          await global.db.collection("collection").doc(id).set(data);
          console.log("WRITE FOR: "+id)
          console.log("COUNT: "+counter++);
      }
    }));
  }
  catch(err) {
    console.log("WRITING TO DB ERROR: "+err);
  }   

}

1. 数据库查询停顿的原因 #

datas.map(async (data) => {...})会立即遍历所有数据项，创建并返回所有异步函数的Promise对象——这就是所有“MAPPING DATA”日志瞬间全部打印的原因。但这些异步函数内部的await db.get()会暂停当前函数执行，直到查询完成。

你看到的停顿本质是数据库SDK的并发请求限制：大部分数据库SDK（包括Firebase）都会对并发请求数做限制，避免服务过载。一次性发起10K个查询请求时，超出限制的请求会被放入队列排队，只有前面的请求完成后，后面的才会开始执行，看起来就像所有get()都在等待，出现了“停顿”。

2. 写入操作无明显阻塞的原因 #

写入操作的await db.set()同样会暂停当前异步函数，但两个因素导致它看起来没有阻塞：

• 部分数据库的写入操作会做批量优化，或者SDK对写入的并发限制更宽松；
• 不同写入请求的完成时间有差异，当部分写入完成后，对应的console.log("WRITE FOR: "+id)就会输出，所以呈现持续输出的状态，而非像查询那样集中在后期批量输出。

3. 实现“并行+单数据项顺序执行”的方案 #

要保持每个数据项内部“先查再写”的顺序，同时尽可能并行处理不同数据项，核心是控制并发数，避免一次性发起过多请求触发SDK的队列限制。以下是两种可行方案：

方案一：分批并行处理 #

把10K条数据分成若干批次（比如每批100条），先并行处理完一批，再处理下一批。既控制了并发数，又能保证批次内的并行执行：

async function batchParallelWrites(datas, batchSize = 100) {
  try {
    let counter = 1;
    // 分割数据为批次
    const batches = [];
    for (let i = 0; i < datas.length; i += batchSize) {
      batches.push(datas.slice(i, i + batchSize));
    }

    // 逐批处理
    for (const batch of batches) {
      await Promise.all(batch.map(async (data) => {
        const id = data.id; // 假设data包含id字段
        console.log(`PROCESSING: ${counter++}/${datas.length}`);
        // 先检查数据库
        const snap = await global.db.collection("Prompts").doc(id).get();
        if (!snap.data()) {
          // 按需写入
          await global.db.collection("collection").doc(id).set(data);
          console.log(`WRITE FOR: ${id}`);
        } else {
          console.log(`NO WRITE FOR: ${id}`);
        }
      }));
    }
  } catch (err) {
    console.log(`WRITING TO DB ERROR: ${err}`);
  }
}

方案二：并发池控制（更高效） #

用自定义并发池控制同时执行的异步任务数，避免分批的等待时间，让系统保持稳定的并发量：

async function poolParallelWrites(datas, concurrency = 100) {
  try {
    let counter = 1;
    const pool = [];
    const executing = new Set();

    for (const data of datas) {
      const id = data.id;
      // 创建异步任务
      const task = async () => {
        console.log(`PROCESSING: ${counter++}/${datas.length}`);
        const snap = await global.db.collection("Prompts").doc(id).get();
        if (!snap.data()) {
          await global.db.collection("collection").doc(id).set(data);
          console.log(`WRITE FOR: ${id}`);
        } else {
          console.log(`NO WRITE FOR: ${id}`);
        }
      };

      pool.push(task());
      executing.add(task());
      // 并发数超过限制时，等待一个任务完成再继续
      if (executing.size >= concurrency) {
        await Promise.race(executing);
        executing.delete([...executing][0]);
      }
    }

    // 等待所有剩余任务完成
    await Promise.all(pool);
  } catch (err) {
    console.log(`WRITING TO DB ERROR: ${err}`);
  }
}

额外优化建议 #

• 修正日志计数器：原代码中的mapped、dbcheck计数器会因异步执行顺序混乱，导致日志数字不连续，建议改用单个counter跟踪处理进度；
• 使用批量写入API：如果数据库支持批量写入（比如Firebase的WriteBatch），可以把需要写入的操作收集起来批量提交，大幅提升性能并减少请求数；
• 细化错误处理：在单个任务内部捕获错误，避免一个任务失败导致整个Promise.all终止。

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