MongoDB 4.2执行大量事务时如何避免“Transaction 1已提交”错误？

先给你拆解这个错误的核心含义：当你尝试在一个已经提交的事务会话中执行数据库操作时，就会抛出这个错误。你的90%操作成功、剩余失败，正是因为这些失败的操作是在事务已经提交之后才开始执行的。

问题根源：forEach不支持异步函数 #

看你的代码，find().forEach(async row => { ... })这里是致命的时序问题：JavaScript的forEach方法并不会等待异步回调函数执行完成。也就是说，你的代码会跳过所有updateOne的等待逻辑，直接走到await session.commitTransaction()这一步——此时事务已经提交，但还有大量updateOne操作还在排队等待执行，这些晚到的操作自然会因为事务已提交而报错。

解决方案1：用for...of替代forEach #

for...of循环可以正确等待每个异步操作完成，确保所有更新都在事务提交前执行完毕：

const rowCollection = client.db('docemur').collection('row')
session.startTransaction()

// 先把查询结果转成数组
const rows = await rowCollection
  .find({ table: table._id }, { session })
  .toArray()

// 用for...of遍历，确保每个异步操作等待完成
for (const row of rows) {
  // 对行数组进行重排序
  row.row = sortBy(row.row, unsorted => {
    return findIndex(columns, sorted => unsorted.column === sorted.identifier)
  })
  // 更新已排序的行
  await rowCollection
    .updateOne(
      { _id: row._id },
      { $set: { row: row.row } },
      { session }
    )
}

await session.commitTransaction()
session.endSession()

解决方案2：用bulkWrite批量更新（更推荐） #

5万次单条updateOne效率极低，还容易因为事务执行时间过长触发超时问题。换成bulkWrite可以把所有更新合并成一次请求，大幅提升性能，同时彻底避免异步循环的时序问题：

const rowCollection = client.db('docemur').collection('row')
session.startTransaction()

const rows = await rowCollection
  .find({ table: table._id }, { session })
  .toArray()

// 构建批量更新操作数组
const bulkOperations = rows.map(row => {
  const sortedRow = sortBy(row.row, unsorted => {
    return findIndex(columns, sorted => unsorted.column === sorted.identifier)
  })
  return {
    updateOne: {
      filter: { _id: row._id },
      update: { $set: { row: sortedRow } }
    }
  }
})

// 执行批量更新
await rowCollection.bulkWrite(bulkOperations, { session })

await session.commitTransaction()
session.endSession()

额外优化建议 #

• 调整事务超时时间：MongoDB默认事务超时是60秒，5万次操作（哪怕是批量）可能超过这个时间，启动会话时可以设置更长的超时：

  const session = client.startSession({
    defaultTransactionOptions: {
      maxCommitTimeMS: 300000 // 设置为5分钟，可根据实际情况调整
    }
  })
  

• 检查oplog大小：虽然你升级到了4.2，但如果oplog容量不足，事务执行过程中可能出现同步问题。可以在MongoShell中执行db.getReplicationInfo()查看oplog状态，必要时调整oplog大小。

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