Firestore云函数每秒读数据队列更新的最优方案及去重解决

核心问题根源 #

1. 重复触发引发并发更新：你用Pub/Sub每分钟触发云函数，内部又嵌套每秒轮询的cron，相当于每分钟会启动60次查询和更新流程，同一文档会被多个并行进程读取并更新。如果newStat包含动态生成的时间戳，arrayUnion会把这些带不同时间戳的对象当成不同元素重复添加。
2. 过滤逻辑错误：itemQueue.filter((val) => val.stat[itemQueue.stat.length - 1].currentStat == 'primary')写法错误，应该取当前元素val的stat数组最后一项，而非itemQueue的。
3. 异步处理混乱：forEach里嵌套async/await不会等待异步操作完成，导致多个更新请求同时发向同一个文档，进一步加剧重复更新问题。
4. 冗余查询操作：已经从collectionGroup拿到目标文档，没必要再通过uid重新查询一次，完全是多余操作。

优化实现方案 #

直接用Pub/Sub调度控制触发频率（如果需要每秒执行就改成* * * * * *，注意Firestore配额限制），结合事务确保更新的原子性，彻底避免重复操作：

const autoUpdate = functions.pubsub.schedule('* * * * *') // 按需调整触发间隔，每秒执行则改为* * * * * *
  .onRun(async (context) => {
    const nowDate = new Date();
    // 定义要添加的新状态，若需时间戳确保唯一性，后续通过事务控制只添加一次
    const newStat = {
      currentStat: 'your-target-status',
      updatedAt: nowDate
      // 其他必填字段...
    };

    // 1. 查询符合条件的文档：endDate为null
    const querySnapshot = await db.collectionGroup('products')
      .where('endDate', '==', null)
      .get();

    const updatePromises = [];

    // 2. 遍历文档，用事务做原子性更新
    for (const doc of querySnapshot.docs) {
      const data = doc.data();
      // 内存中先做一次条件校验：stat最后一项为primary
      if (!data.stat || data.stat.length === 0) continue;
      const lastStat = data.stat[data.stat.length - 1];
      if (lastStat.currentStat !== 'primary') continue;

      // 事务内再次校验并执行更新，避免并发修改
      const updatePromise = db.runTransaction(async (tx) => {
        const docRef = doc.ref;
        const latestSnap = await tx.get(docRef);
        if (!latestSnap.exists) return;
        
        const latestData = latestSnap.data();
        // 二次校验：确保文档状态未被其他进程修改
        if (latestData.endDate !== null) return;
        if (!latestData.stat || latestData.stat.length === 0) return;
        const latestLastStat = latestData.stat[latestData.stat.length - 1];
        if (latestLastStat.currentStat !== 'primary') return;

        // 执行更新操作
        tx.update(docRef, {
          endDate: nowDate,
          stat: admin.firestore.FieldValue.arrayUnion(newStat)
        });
      });

      updatePromises.push(updatePromise);
    }

    // 等待所有事务完成
    await Promise.all(updatePromises);
    functions.logger.log(`处理完成，共更新${updatePromises.length}个文档`);
    return null;
  });

关键优化点 #

• 移除嵌套cron：直接通过Pub/Sub调度控制触发频率，避免重复启动多个并行处理流程。
• 事务双重校验：在事务内部重新读取文档并校验条件，确保只有当文档依然符合endDate为null且最后状态是primary时才会更新，彻底杜绝并发修改导致的重复添加。
• 修正过滤逻辑：内存中先过滤一次，事务内再做二次校验，双重保障准确性。
• 避免冗余操作：直接使用查询到的docRef，不再做多余的uid查询。

额外注意事项 #

• 如果newStat必须包含动态唯一值（比如不同时间戳），依赖事务的原子性就能确保每个文档只添加一次，无需修改newStat结构。
• 注意Firestore的事务配额和写入限制，避免短时间内触发大量更新导致配额超限。

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