使用 MongoDB Node.js 驱动执行事务-未完纪

概述

本文介绍如何通过 MongoDB Node.js 驱动执行事务。事务可以包含一系列操作，在整个事务提交前，这些操作不会让数据变更对外可见。如果任意操作失败，驱动会结束事务，并在变更可见之前丢弃全部修改。这种特性称为原子性。

MongoDB 对单个文档的全部写操作本来就是原子的。如果需要原子地修改多个文档，则需要多文档事务。多文档事务满足 ACID：即使驱动遇到意外错误，MongoDB 也能保证事务涉及的数据保持一致。更多信息见 ACID 事务介绍。

注意：执行多文档事务要求连接 MongoDB Server 4.0 或更高版本。详细限制见服务器手册的事务与操作。

下文依次介绍因果一致性、事务 API、事务选项和事务错误。

因果一致性

MongoDB 会在某些客户端会话中启用因果一致性。在分布式系统中，它保证会话内操作按因果顺序执行，客户端观察到的结果符合操作间的依赖关系。例如，一次操作在逻辑上依赖另一次操作的结果，后续读取就应反映这种关系。

要保证因果一致性，会话必须满足：

启动会话时启用因果一致性选项；该选项默认启用。
操作在同一线程的同一会话中运行。否则，会话或线程之间必须传递操作时间与集群时间。两个会话传递这些值的示例见服务器手册的因果一致性示例。
使用 majority 读关注。
使用 majority 写关注，这也是默认写关注值。

保证	说明
读取自己的写入	读取反映此前写操作的结果。
单调读取	后续读取不会返回比先前读取更早的数据状态。
单调写入	必须先执行的写入先执行。例如先 insertOne()，再 updateOne() 修改刚插入的文档，服务器先执行插入。
写入跟随读取	依赖读取的写入在读取之后执行。例如先 findOne() 获取文档，再 deleteOne() 删除它，服务器先执行查找。

进一步概念说明见服务器手册中的因果一致性与因果一致性与读写关注。

事务 API

驱动提供 Core API 和 Convenient Transaction API 两种接口。

Core API 提供创建、提交和结束事务的基本框架，使用者必须显式创建、提交和结束事务，创建和结束运行事务的会话，并实现错误处理。

便捷事务 API 自动负责提交、结束事务，并包含错误处理逻辑，在服务器返回某些错误时自动重试。具体行为见下文“事务错误”。

重要：连接 MongoDB Server 4.2 或更早版本时，事务只能向已存在的集合写入。连接 4.4 及以后版本时，服务器可在事务写入过程中按需自动创建集合。详见在事务中创建集合和索引。

Core API

它提供以下方法：

startSession()：创建新的 ClientSession。
startTransaction()：开始新事务。
commitTransaction()：在创建事务的会话中提交当前事务。
abortTransaction()：在创建事务的会话中中止当前事务。
endSession()：结束当前会话。

每个需要在会话中执行的操作，都必须传入会话实例；同时实现 catch 块，识别服务器事务错误并执行错误处理。

示例：

async function coreTest(client) {
  const session = client.startSession();
  try {
    session.startTransaction();

    const savingsColl = client.db("bank").collection("savings_accounts");
    await savingsColl.findOneAndUpdate(
      {account_id: "9876"}, 
      {$inc: {amount: -100 }}, 
      { session });

    const checkingColl = client.db("bank").collection("checking_accounts");
    await checkingColl.findOneAndUpdate(
      {account_id: "9876"}, 
      {$inc: {amount: 100 }}, 
      { session });

    // ... perform other operations

    await session.commitTransaction();
    console.log("Transaction committed.");
  } catch (error) {
    console.log("An error occurred during the transaction:" + error);
    await session.abortTransaction();
  } finally {
    await session.endSession();
  }
}

重要：会话必须与创建它的客户端一起使用。把一个 MongoClient 创建的会话交给另一个客户端，会导致错误。下面由 client1 创建 ClientSession，却传给 client2 执行写入，因此产生 MongoInvalidArgumentError：

const session = client1.startSession();
client2.db('myDB').collection('myColl').insertOne({ name: 'Jane Eyre' }, { session });

显式资源管理：Node.js 驱动原生支持 MongoClient、ClientSession、ChangeStreams 与游标的显式资源管理。该功能仍为实验性，可能变化。用法见 v6.9 发行说明。

完整可运行示例见使用 Core API。

便捷事务 API

该 API 提供：

withSession()：在会话中运行回调，自动创建并结束会话。
withTransaction()：在事务中运行回调，回调返回时调用 commitTransaction()。

这两个方法返回回调的返回值。例如，传给 withTransaction() 的回调返回 { hello: "world" }，该方法也返回这个文档。

重要：为避免无限循环错误，应确保传给 withTransaction() 的回调捕获自身抛出的错误。

可以将回调返回值通过 withTransaction() 和 withSession() 逐层返回，以便在其他代码中使用。使用时必须：

将会话实例传给每个会话内操作。
对每个操作使用 async／await。
避免并行，例如不要调用 Promise.all()。并行使用会话通常会导致服务器错误。

示例：

async function convTest(client) {
  let txnRes = await client.withSession(async (session) =>
    session.withTransaction(async (session) => {
      const savingsColl = client.db("bank").collection("savings_accounts");
      await savingsColl.findOneAndUpdate(
        {account_id: "9876"}, 
        {$inc: {amount: -100 }}, 
        { session });
  
      const checkingColl = client.db("bank").collection("checking_accounts");
      await checkingColl.findOneAndUpdate(
        {account_id: "9876"}, 
        {$inc: {amount: 100 }}, 
        { session });

      // ... perform other operations

      return "Transaction committed.";
    }, null)
  );
  console.log(txnRes);
}

完整可运行示例见使用便捷事务 API。

注意：Node.js 驱动不支持在单个事务内并行运行操作。

事务选项

可以向 startTransaction() 或 withTransaction() 传入 TransactionOptions，配置事务行为。显式指定的值会覆盖 MongoClient 上对应设置。

设置	说明
readConcern	指定副本集读取的一致性，详见服务器手册的 Read Concern。
writeConcern	指定要求副本集确认写入的级别，详见 Write Concern。
readPreference	指定如何将读取路由到副本集成员，详见 Read Preference。
maxCommitTimeMS	事务提交操作允许运行的最长时间，单位毫秒。

完整选项列表见 TransactionOptions API 文档。若事务选项没有指定，事务会继承 MongoClient 的设置。

下面定义选项并传给 startTransaction()：

const txnOpts = {
  readPreference: 'primary',
  readConcern: { level: 'local' },
  writeConcern: { w: 'majority' },
  maxCommitTimeMS: 1000
};
session.startTransaction(txnOpts);