MongoDB 无匹配时如何并发安全地 insert many

2026-10-24支付阿杰

结论

不要用“先查询、再插入”的方式实现幂等，应该让 MongoDB 直接约束重复写入：

为同一批 token 中的每条记录分配稳定序号，例如 tokenIndex。
为 paymentIntentId + tokenIndex 创建唯一复合索引。
通过 upsert 和 $setOnInsert 批量写入。

即使多个请求同时为同一笔支付创建 4 个 token，数据库最终也只会留下 tokenIndex 为 0、1、2、3 的四条记录。

为什么需要 tokenIndex

现有数据结构无法区分某个 token 是一笔支付购买的第几个。如果只为 origin.paymentIntentId 创建唯一索引，每笔支付就只能对应一个 token，不符合需求。

因此，每个 token 都需要一个可以重复计算的稳定标识：

{
  _id: ObjectId,
  owner: ObjectId,
  type: "standard",
  tokenIndex: 0,
  origin: {
    paymentIntentId: "pi_123",
    date: ISODate("2026-09-01T00:00:00Z")
  }
}

一笔支付购买 4 个 token 时，tokenIndex 始终是 0 到 3。请求重试时也必须生成相同的序号，不能改用随机值。

建立唯一索引

如果 paymentIntentId 在整个系统内全局唯一，可以创建以下索引：

db.tokens.createIndex(
  {
    "origin.paymentIntentId": 1,
    tokenIndex: 1
  },
  {
    unique: true,
    name: "uniq_payment_intent_token_index"
  }
)

如果同一个 paymentIntentId 可能出现在不同的支付提供方、账户或测试环境中，还要把对应的作用域字段加入唯一索引。例如：

db.tokens.createIndex(
  {
    paymentProvider: 1,
    paymentAccountId: 1,
    "origin.paymentIntentId": 1,
    tokenIndex: 1
  },
  {
    unique: true
  }
)

是否将 owner 和 type 放进索引，要看业务如何定义幂等键。通常，支付完成后，token 的归属和类型不应因请求重试而改变。更合适的做法是用 paymentIntentId + tokenIndex 唯一标识一个 token，再单独检查其他参数是否一致。

使用 bulkWrite 和 upsert

以下示例使用 MongoDB Node.js Driver：

import { ObjectId } from "mongodb";

async function issueTokens({
  tokensCollection,
  owner,
  type,
  paymentIntentId,
  quantity,
  paymentDate
}) {
  if (!Number.isSafeInteger(quantity) || quantity <= 0) {
    throw new Error("quantity must be a positive integer");
  }

  const operations = Array.from({ length: quantity }, (_, tokenIndex) => ({
    updateOne: {
      filter: {
        "origin.paymentIntentId": paymentIntentId,
        tokenIndex
      },
      update: {
        $setOnInsert: {
          owner: new ObjectId(owner),
          type,
          tokenIndex,
          origin: {
            paymentIntentId,
            date: paymentDate
          }
        }
      },
      upsert: true
    }
  }));

  return tokensCollection.bulkWrite(operations, {
    ordered: false
  });
}

每个 updateOne 会按以下方式处理：

对应的 token 已存在时，操作会匹配现有记录，但不会执行 $setOnInsert，因此不会修改记录。
对应的 token 不存在时，操作会插入一条新记录。
多个请求并发执行时，唯一索引会阻止相同的 paymentIntentId + tokenIndex 被插入两次。

设置 ordered: false 后，批量操作可以彼此独立执行。即使其中一条发生冲突，其他序号仍可继续处理。

在极端并发情况下，受驱动程序或 MongoDB 版本的 upsert 执行路径影响，某个参与竞争的请求仍可能收到 E11000 duplicate key。如果发生冲突的确实是上述幂等唯一索引，这通常说明另一个请求已经完成了相同写入。此时可以重新执行同一组 upsert，或者读取并验证最终记录。不要不加区分地忽略所有重复键错误。

验证重试参数是否一致

$setOnInsert 会保留首次成功请求写入的 owner、type 和 date。如果后续请求使用相同的 paymentIntentId，却传入不同的用户、类型或数量，简单地跳过写入可能会掩盖业务错误。

可以在写入后查询并校验：

const existingTokens = await tokensCollection
  .find({
    "origin.paymentIntentId": paymentIntentId
  })
  .sort({ tokenIndex: 1 })
  .toArray();

if (existingTokens.length !== quantity) {
  throw new Error("paymentIntentId was reused with a different quantity");
}

const hasConflict = existingTokens.some(
  token =>
    !token.owner.equals(new ObjectId(owner)) ||
    token.type !== type
);

if (hasConflict) {
  throw new Error("paymentIntentId was reused with different token parameters");
}

这次查询只负责检查业务参数是否一致，不用于保证并发安全。并发约束仍然由唯一索引负责。

如果要求“四条全部写入，或者一条都不写”

上述 upsert 方案能保证数据库最终收敛为四条唯一 token，但不具备“整批原子写入”语义。如果服务写入第二条后崩溃，数据库中可能暂时只有两条记录，下一次重试再补齐剩余记录。

如果业务要求一次操作中的所有记录必须全部成功或全部回滚，就需要使用事务，并增加一个发放记录集合：

db.tokenIssuances.createIndex(
  { paymentIntentId: 1 },
  { unique: true }
)

在事务中，先插入一条具有唯一约束的发放记录，再插入全部 token：

async function issueTokensAtomically({
  client,
  issuancesCollection,
  tokensCollection,
  owner,
  type,
  paymentIntentId,
  quantity,
  paymentDate
}) {
  const session = client.startSession();

  try {
    await session.withTransaction(async () => {
      await issuancesCollection.insertOne(
        {
          paymentIntentId,
          owner: new ObjectId(owner),
          type,
          quantity,
          createdAt: new Date()
        },
        { session }
      );

      const tokens = Array.from({ length: quantity }, (_, tokenIndex) => ({
        owner: new ObjectId(owner),
        type,
        tokenIndex,
        origin: {
          paymentIntentId,
          date: paymentDate
        }
      }));

      await tokensCollection.insertMany(tokens, { session });
    });
  } catch (error) {
    if (error?.code === 11000) {
      // 必须确认冲突来自 paymentIntentId 唯一索引，
      // 然后读取已有 issuance 并校验 owner、type 和 quantity。
      return;
    }

    throw error;
  } finally {
    await session.endSession();
  }
}

事务方案适用于必须保证“整批只发放一次”的场景，但 MongoDB 部署需要支持事务。如果 token 发放允许安全重试，也能接受短时间内只完成部分写入，那么唯一索引配合幂等 upsert 通常更简单，失败后也更容易恢复。

注意事项

并发安全依赖唯一索引，应用层查询无法替代这一约束。
幂等键必须使用可信、稳定的支付标识，不能在每次请求时随机生成。
创建唯一索引前，需要先检查并清理已有的重复数据，否则索引可能创建失败。
不要看到错误码 11000 就直接忽略异常，必须确认冲突来自预期的幂等索引。
如果支付数量可能发生变化，需要明确处理规则，例如拒绝修改、补发差额或创建新的支付版本。不要无条件允许同一 paymentIntentId 对应不同数量。
如果还要同步更新余额、账本或订单状态，应将这些写入放入同一个事务，或者交给可重试的状态机处理。