认识 MongoDB 4.0 的新特性——事务（Transactions）

时间 2019-11-29

标签认识 mongodb 4.0 特性事务 transactions 栏目 MongoDB 繁體版

原文原文链接

前言

相信使用过主流的关系型数据库的朋友对“事务（Transactions）”不会太陌生，它可让咱们把对多张表的屡次数据库操做整合为一次原子操做，这在高并发场景下能够保证多个数据操做之间的互不干扰；而且一旦在这些操做过程任一环节中出现了错误，事务会停止而且让数据回滚，这使得同时在多张表中修改数据的时候保证了数据的一致性。javascript

之前 MongoDB 是不支持事务的，所以开发者在须要用到事务的时候，不得不借用其余工具，在业务代码层面去弥补数据库的不足。随着 4.0 版本的发布，MongoDB 也为咱们带来了原生的事务操做，下面就让咱们一块儿来认识它，并经过简单的例子了解如何去使用。html

介绍

事务和副本集（Replica Sets）

副本集是 MongoDB 的一种主副节点架构，它使数据获得最大的可用性，避免单点故障引发的整个服务不能访问的状况的发生。目前 MongoDB 的多表事务操做仅支持在副本集上运行，想要在本地环境安装运行副本集能够借助一个工具包——run-rs，如下的文章中有详细的使用说明：前端

thecodebarbarian.com/introducing…java

事务和会话（Sessions）

事务和会话（Sessions）关联，一个会话同一时刻只能开启一个事务操做，当一个会话断开，这个会话中的事务也会结束。git

事务中的函数

Session.startTransaction()

在当前会话中开始一次事务，事务开启后就能够开始进行数据操做。在事务中执行的数据操做是对外隔离的，也就是说事务中的操做是原子性的。github

Session.commitTransaction()

提交事务，将事务中对数据的修改进行保存，而后结束当前事务，一次事务在提交以前的数据操做对外都是不可见的。redis

Session.abortTransaction()

停止当前的事务，并将事务中执行过的数据修改回滚。mongodb

重试

当事务运行中报错，catch 到的错误对象中会包含一个属性名为 errorLabels 的数组，当这个数组中包含如下2个元素的时候，表明咱们能够从新发起相应的事务操做。数据库

TransientTransactionError：出如今事务开启以及随后的数据操做阶段
UnknownTransactionCommitResult：出如今提交事务阶段

示例

通过上面的铺垫，你是否是已经火烧眉毛想知道究竟应该怎么写代码去完成一次完整的事务操做？下面咱们就简单写一个例子：json

场景描述： 假设一个交易系统中有2张表——记录商品的名称、库存数量等信息的表 commodities，和记录订单的表 orders。当用户下单的时候，首先要找到 commodities 表中对应的商品，判断库存数量是否知足该笔订单的需求，是的话则减去相应的值，而后在 orders 表中插入一条订单数据。在高并发场景下，可能在查询库存数量和减小库存的过程当中，又收到了一次新的建立订单请求，这个时候可能就会出问题，由于新的请求在查询库存的时候，上一次操做还未完成减小库存的操做，这个时候查询到的库存数量多是充足的，因而开始执行后续的操做，实际上可能上一次操做减小了库存后，库存的数量就已经不足了，因而新的下单请求可能就会致使实际建立的订单数量超过库存数量。

以往要解决这个问题，咱们能够用给商品数据“加锁”的方式，好比基于 Redis 的各类锁，同一时刻只容许一个订单操做一个商品数据，这种方案能解决问题，缺点就是代码更复杂了，而且性能会比较低。若是用数据库事务的方式就能够简洁不少：

commodities 表数据（stock 为库存）：

{ "_id" : ObjectId("5af0776263426f87dd69319a"), "name" : "灭霸原味手套", "stock" : 5 }
{ "_id" : ObjectId("5af0776263426f87dd693198"), "name" : "雷神专用铁锤", "stock" : 2 }
复制代码

orders 表数据：

{ "_id" : ObjectId("5af07daa051d92f02462644c"), "commodity": ObjectId("5af0776263426f87dd69319a"), "amount": 2 }
{ "_id" : ObjectId("5af07daa051d92f02462644b"), "commodity": ObjectId("5af0776263426f87dd693198"), "amount": 3 }
复制代码

经过一次事务完成建立订单操做（mongo Shell）：

// 执行 txnFunc 而且在遇到 TransientTransactionError 的时候重试
function runTransactionWithRetry(txnFunc, session) {
  while (true) {
    try {
      txnFunc(session); // 执行事务
      break;
    } catch (error) {
      if (
        error.hasOwnProperty('errorLabels') &&
        error.errorLabels.includes('TransientTransactionError')
      ) {
        print('TransientTransactionError, retrying transaction ...');
        continue;
      } else {
        throw error;
      }
    }
  }
}

// 提交事务而且在遇到 UnknownTransactionCommitResult 的时候重试
function commitWithRetry(session) {
  while (true) {
    try {
      session.commitTransaction();
      print('Transaction committed.');
      break;
    } catch (error) {
      if (
        error.hasOwnProperty('errorLabels') &&
        error.errorLabels.includes('UnknownTransactionCommitResult')
      ) {
        print('UnknownTransactionCommitResult, retrying commit operation ...');
        continue;
      } else {
        print('Error during commit ...');
        throw error;
      }
    }
  }
}

// 在一次事务中完成建立订单操做
function createOrder(session) {
  var commoditiesCollection = session.getDatabase('mall').commodities;
  var ordersCollection = session.getDatabase('mall').orders;
  // 假设该笔订单中商品的数量
  var orderAmount = 3;
  // 假设商品的ID
  var commodityID = ObjectId('5af0776263426f87dd69319a');

  session.startTransaction({
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' },
  });

  try {
    var { stock } = commoditiesCollection.findOne({ _id: commodityID });
    if (stock < orderAmount) {
      print('Stock is not enough');
      session.abortTransaction();
      throw new Error('Stock is not enough');
    }
    commoditiesCollection.updateOne(
      { _id: commodityID },
      { $inc: { stock: -orderAmount } }
    );
    ordersCollection.insertOne({
      commodity: commodityID,
      amount: orderAmount,
    });
  } catch (error) {
    print('Caught exception during transaction, aborting.');
    session.abortTransaction();
    throw error;
  }

  commitWithRetry(session);
}

// 发起一次会话
var session = db.getMongo().startSession({ readPreference: { mode: 'primary' } });

try {
  runTransactionWithRetry(createOrder, session);
} catch (error) {
  // 错误处理
} finally {
  session.endSession();
}
复制代码

上面的代码看着感受不少，其实 runTransactionWithRetry 和 commitWithRetry 这两个函数都是能够抽离出来成为公共函数的，不须要每次操做都重复书写。用上了事务以后，由于事务中的数据操做都是一次原子操做，因此咱们就不须要考虑分布并发致使的数据一致性的问题，是否是感受简单了许多？

你可能注意到了，代码中在执行 startTransaction 的时候设置了两个参数——readConcern 和 writeConcern，这是 MongoDB 读写操做的确认级别，在这里用于在副本集中平衡数据读写操做的可靠性和性能，若是在这里展开就太多了，因此感兴趣的朋友建议去阅读官方文档了解一下：

readConcern：

docs.mongodb.com/master/refe…

writeConcern：

docs.mongodb.com/master/refe…