Mongoose 中的事务

事务允许您以隔离的方式执行多个操作，如果其中一个操作失败，则可以撤销所有操作。本指南将帮助您开始使用 Mongoose 中的事务。

事务入门

如果您还没有，请导入 mongoose

import mongoose from 'mongoose';

要创建事务，您首先需要使用 Mongoose#startSession 或 Connection#startSession() 创建一个会话。

// Using Mongoose's default connection
const session = await mongoose.startSession();

// Using custom connection
const db = await mongoose.createConnection(mongodbUri).asPromise();
const session = await db.startSession();

在实践中，您应该使用 session.withTransaction() 辅助函数或 Mongoose 的 Connection#transaction() 函数来运行事务。session.withTransaction() 辅助函数处理

创建事务
如果操作成功，则提交事务
如果您的操作抛出异常，则中止事务
在出现瞬态事务错误时重试。

let session = null;
return Customer.createCollection().
  then(() => Customer.startSession()).
  // The `withTransaction()` function's first parameter is a function
  // that returns a promise.
  then(_session => {
    session = _session;
    return session.withTransaction(() => {
      return Customer.create([{ name: 'Test' }], { session: session });
    });
  }).
  then(() => Customer.countDocuments()).
  then(count => assert.strictEqual(count, 1)).
  then(() => session.endSession());

有关 ClientSession#withTransaction() 函数的更多信息，请参阅 MongoDB Node.js 驱动程序文档。

Mongoose 的 Connection#transaction() 函数是对 withTransaction() 的包装，它将 Mongoose 更改跟踪与事务集成。例如，假设您在一个后来失败的事务中 save() 了一个文档。该文档中的更改不会持久保存到 MongoDB。Connection#transaction() 函数通知 Mongoose 更改跟踪 save() 已回滚，并将事务中更改的所有字段标记为已修改。

const doc = new Person({ name: 'Will Riker' });

await db.transaction(async function setRank(session) {
  doc.name = 'Captain';
  await doc.save({ session });
  doc.isNew; // false

  // Throw an error to abort the transaction
  throw new Error('Oops!');
}, { readPreference: 'primary' }).catch(() => {});

// true, `transaction()` reset the document's state because the
// transaction was aborted.
doc.isNew;

关于事务中并行性的说明

在事务期间，**不支持** 并行运行操作。使用 Promise.all、Promise.allSettled、Promise.race 等来并行化事务内的操作是未定义的行为，应该避免。

与 Mongoose 文档和 `save()` 一起使用

如果您从使用会话的 Mongoose 文档中获取 findOne() 或 find()，则该文档将保留对会话的引用，并使用该会话进行 save()。

要获取/设置与给定文档关联的会话，请使用 doc.$session()。

const User = db.model('User', new Schema({ name: String }));

let session = null;
return User.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    return User.create({ name: 'foo' });
  }).
  then(() => {
    session.startTransaction();
    return User.findOne({ name: 'foo' }).session(session);
  }).
  then(user => {
    // Getter/setter for the session associated with this document.
    assert.ok(user.$session());
    user.name = 'bar';
    // By default, `save()` uses the associated session
    return user.save();
  }).
  then(() => User.findOne({ name: 'bar' })).
  // Won't find the doc because `save()` is part of an uncommitted transaction
  then(doc => assert.ok(!doc)).
  then(() => session.commitTransaction()).
  then(() => session.endSession()).
  then(() => User.findOne({ name: 'bar' })).
  then(doc => assert.ok(doc));

使用聚合框架

Model.aggregate() 函数也支持事务。Mongoose 聚合有一个 session() 辅助函数，它设置 session 选项。以下是执行事务内聚合的示例。

const Event = db.model('Event', new Schema({ createdAt: Date }), 'Event');

let session = null;
return Event.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    session.startTransaction();
    return Event.insertMany([
      { createdAt: new Date('2018-06-01') },
      { createdAt: new Date('2018-06-02') },
      { createdAt: new Date('2017-06-01') },
      { createdAt: new Date('2017-05-31') }
    ], { session: session });
  }).
  then(() => Event.aggregate([
    {
      $group: {
        _id: {
          month: { $month: '$createdAt' },
          year: { $year: '$createdAt' }
        },
        count: { $sum: 1 }
      }
    },
    { $sort: { count: -1, '_id.year': -1, '_id.month': -1 } }
  ]).session(session)).
  then(res => assert.deepEqual(res, [
    { _id: { month: 6, year: 2018 }, count: 2 },
    { _id: { month: 6, year: 2017 }, count: 1 },
    { _id: { month: 5, year: 2017 }, count: 1 }
  ])).
  then(() => session.commitTransaction()).
  then(() => session.endSession());

使用 AsyncLocalStorage

Mongoose 中事务的一个主要痛点是您需要记住在每个操作上设置 session 选项。如果您没有，您的操作将在事务之外执行。Mongoose 8.4 能够使用 Node 的 AsyncLocalStorage API 在 Connection.prototype.transaction() 执行器函数中的所有操作上设置 session 操作。使用 mongoose.set('transactionAsyncLocalStorage', true) 设置 transactionAsyncLocalStorage 选项以启用此功能。

mongoose.set('transactionAsyncLocalStorage', true);

const Test = mongoose.model('Test', mongoose.Schema({ name: String }));

const doc = new Test({ name: 'test' });

// Save a new doc in a transaction that aborts
await connection.transaction(async() => {
  await doc.save(); // Notice no session here
  throw new Error('Oops');
}).catch(() => {});

// false, `save()` was rolled back
await Test.exists({ _id: doc._id });

使用 transactionAsyncLocalStorage，您不再需要将会话传递给每个操作。Mongoose 将在幕后默认添加会话。

高级用法

希望对何时提交或中止事务有更精细控制的高级用户可以使用 session.startTransaction() 启动事务

const Customer = db.model('Customer', new Schema({ name: String }));

let session = null;
return Customer.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    // Start a transaction
    session.startTransaction();
    // This `create()` is part of the transaction because of the `session`
    // option.
    return Customer.create([{ name: 'Test' }], { session: session });
  }).
  // Transactions execute in isolation, so unless you pass a `session`
  // to `findOne()` you won't see the document until the transaction
  // is committed.
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(!doc)).
  // This `findOne()` will return the doc, because passing the `session`
  // means this `findOne()` will run as part of the transaction.
  then(() => Customer.findOne({ name: 'Test' }).session(session)).
  then(doc => assert.ok(doc)).
  // Once the transaction is committed, the write operation becomes
  // visible outside of the transaction.
  then(() => session.commitTransaction()).
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(doc)).
  then(() => session.endSession());

您也可以使用 session.abortTransaction() 中止事务

let session = null;
return Customer.createCollection().
  then(() => Customer.startSession()).
  then(_session => {
    session = _session;
    session.startTransaction();
    return Customer.create([{ name: 'Test' }], { session: session });
  }).
  then(() => Customer.create([{ name: 'Test2' }], { session: session })).
  then(() => session.abortTransaction()).
  then(() => Customer.countDocuments()).
  then(count => assert.strictEqual(count, 0)).
  then(() => session.endSession());