前言

相信使用過(guò)主流的關(guān)系型數(shù)據(jù)庫(kù)的朋友對(duì)“事務(wù)(Transactions)”不會(huì)太陌生，它可以讓我們把對(duì)多張表的多次數(shù)據(jù)庫(kù)操作整合為一次原子操作，這在高并發(fā)場(chǎng)景下可以保證多個(gè)數(shù)據(jù)操作之間的互不干擾;并且一旦在這些操作過(guò)程任一環(huán)節(jié)中出現(xiàn)了錯(cuò)誤，事務(wù)會(huì)中止并且讓數(shù)據(jù)回滾，這使得同時(shí)在多張表中修改數(shù)據(jù)的時(shí)候保證了數(shù)據(jù)的一致性。

以前 MongoDB 是不支持事務(wù)的，因此開發(fā)者在需要用到事務(wù)的時(shí)候，不得不借用其他工具，在業(yè)務(wù)代碼層面去彌補(bǔ)數(shù)據(jù)庫(kù)的不足。隨著 4.0 版本的發(fā)布，MongoDB 也為我們帶來(lái)了原生的事務(wù)操作，下面就讓我們一起來(lái)認(rèn)識(shí)它，并通過(guò)簡(jiǎn)單的例子了解如何去使用。

[[249466]]

介紹

事務(wù)和副本集(Replica Sets)

副本集是 MongoDB 的一種主副節(jié)點(diǎn)架構(gòu)，它使數(shù)據(jù)得到***的可用性，避免單點(diǎn)故障引起的整個(gè)服務(wù)不能訪問(wèn)的情況的發(fā)生。目前 MongoDB 的多表事務(wù)操作僅支持在副本集上運(yùn)行，想要在本地環(huán)境安裝運(yùn)行副本集可以借助一個(gè)工具包——run-rs，以下的文章中有詳細(xì)的使用說(shuō)明：

https://thecodebarbarian.com/...

事務(wù)和會(huì)話(Sessions)

事務(wù)和會(huì)話(Sessions)關(guān)聯(lián)，一個(gè)會(huì)話同一時(shí)刻只能開啟一個(gè)事務(wù)操作，當(dāng)一個(gè)會(huì)話斷開，這個(gè)會(huì)話中的事務(wù)也會(huì)結(jié)束。

事務(wù)中的函數(shù)

• Session.startTransaction()

在當(dāng)前會(huì)話中開始一次事務(wù)，事務(wù)開啟后就可以開始進(jìn)行數(shù)據(jù)操作。在事務(wù)中執(zhí)行的數(shù)據(jù)操作是對(duì)外隔離的，也就是說(shuō)事務(wù)中的操作是原子性的。

• Session.commitTransaction()

提交事務(wù)，將事務(wù)中對(duì)數(shù)據(jù)的修改進(jìn)行保存，然后結(jié)束當(dāng)前事務(wù)，一次事務(wù)在提交之前的數(shù)據(jù)操作對(duì)外都是不可見的。

• Session.abortTransaction()

中止當(dāng)前的事務(wù)，并將事務(wù)中執(zhí)行過(guò)的數(shù)據(jù)修改回滾。

重試

當(dāng)事務(wù)運(yùn)行中報(bào)錯(cuò)，catch 到的錯(cuò)誤對(duì)象中會(huì)包含一個(gè)屬性名為 errorLabels 的數(shù)組，當(dāng)這個(gè)數(shù)組中包含以下2個(gè)元素的時(shí)候，代表我們可以重新發(fā)起相應(yīng)的事務(wù)操作。

• TransientTransactionError：出現(xiàn)在事務(wù)開啟以及隨后的數(shù)據(jù)操作階段
• UnknownTransactionCommitResult：出現(xiàn)在提交事務(wù)階段

示例

經(jīng)過(guò)上面的鋪墊，你是不是已經(jīng)迫不及待想知道究竟應(yīng)該怎么寫代碼去完成一次完整的事務(wù)操作?下面我們就簡(jiǎn)單寫一個(gè)例子：

場(chǎng)景描述： 假設(shè)一個(gè)交易系統(tǒng)中有2張表——記錄商品的名稱、庫(kù)存數(shù)量等信息的表 commodities，和記錄訂單的表 orders。當(dāng)用戶下單的時(shí)候，首先要找到 commodities 表中對(duì)應(yīng)的商品，判斷庫(kù)存數(shù)量是否滿足該筆訂單的需求，是的話則減去相應(yīng)的值，然后在 orders 表中插入一條訂單數(shù)據(jù)。在高并發(fā)場(chǎng)景下，可能在查詢庫(kù)存數(shù)量和減少庫(kù)存的過(guò)程中，又收到了一次新的創(chuàng)建訂單請(qǐng)求，這個(gè)時(shí)候可能就會(huì)出問(wèn)題，因?yàn)樾碌恼?qǐng)求在查詢庫(kù)存的時(shí)候，上一次操作還未完成減少庫(kù)存的操作，這個(gè)時(shí)候查詢到的庫(kù)存數(shù)量可能是充足的，于是開始執(zhí)行后續(xù)的操作，實(shí)際上可能上一次操作減少了庫(kù)存后，庫(kù)存的數(shù)量就已經(jīng)不足了，于是新的下單請(qǐng)求可能就會(huì)導(dǎo)致實(shí)際創(chuàng)建的訂單數(shù)量超過(guò)庫(kù)存數(shù)量。

以往要解決這個(gè)問(wèn)題，我們可以用給商品數(shù)據(jù)“加鎖”的方式，比如基于 Redis 的各種鎖，同一時(shí)刻只允許一個(gè)訂單操作一個(gè)商品數(shù)據(jù)，這種方案能解決問(wèn)題，缺點(diǎn)就是代碼更復(fù)雜了，并且性能會(huì)比較低。如果用數(shù)據(jù)庫(kù)事務(wù)的方式就可以簡(jiǎn)潔很多：

commodities 表數(shù)據(jù)(stock 為庫(kù)存)：

{ "_id" : ObjectId("5af0776263426f87dd69319a"), "name" : "滅霸原味手套", "stock" : 5 } 
{ "_id" : ObjectId("5af0776263426f87dd693198"), "name" : "雷神專用鐵錘", "stock" : 2 } 

orders 表數(shù)據(jù)：

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

通過(guò)一次事務(wù)完成創(chuàng)建訂單操作(mongo Shell)：

// 執(zhí)行 txnFunc 并且在遇到 TransientTransactionError 的時(shí)候重試 
function runTransactionWithRetry(txnFunc, session) { 
  while (true) { 
    try { 
      txnFunc(session); // 執(zhí)行事務(wù) 
      break; 
    } catch (error) { 
      if ( 
        error.hasOwnProperty('errorLabels') && 
        error.errorLabels.includes('TransientTransactionError') 
      ) { 
        print('TransientTransactionError, retrying transaction ...'); 
        continue; 
      } else { 
        throw error; 
      } 
    } 
  } 
} 
 
// 提交事務(wù)并且在遇到 UnknownTransactionCommitResult 的時(shí)候重試 
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; 
      } 
    } 
  } 
} 
 
// 在一次事務(wù)中完成創(chuàng)建訂單操作 
function createOrder(session) { 
  var commoditiesCollection = session.getDatabase('mall').commodities; 
  var ordersCollection = session.getDatabase('mall').orders; 
  // 假設(shè)該筆訂單中商品的數(shù)量 
  var orderAmount = 3; 
  // 假設(shè)商品的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); 
} 
 
// 發(fā)起一次會(huì)話 
var session = db.getMongo().startSession({ readPreference: { mode: 'primary' } }); 
 
try { 
  runTransactionWithRetry(createOrder, session); 
} catch (error) { 
  // 錯(cuò)誤處理 
} finally { 
  session.endSession(); 
} 

上面的代碼看著感覺很多，其實(shí) runTransactionWithRetry 和 commitWithRetry 這兩個(gè)函數(shù)都是可以抽離出來(lái)成為公共函數(shù)的，不需要每次操作都重復(fù)書寫。用上了事務(wù)之后，因?yàn)槭聞?wù)中的數(shù)據(jù)操作都是一次原子操作，所以我們就不需要考慮分布并發(fā)導(dǎo)致的數(shù)據(jù)一致性的問(wèn)題，是不是感覺簡(jiǎn)單了許多?

你可能注意到了，代碼中在執(zhí)行 startTransaction 的時(shí)候設(shè)置了兩個(gè)參數(shù)——readConcern 和 writeConcern，這是 MongoDB 讀寫操作的確認(rèn)級(jí)別，在這里用于在副本集中平衡數(shù)據(jù)讀寫操作的可靠性和性能，如果在這里展開就太多了，所以感興趣的朋友建議去閱讀官方文檔了解一下：

readConcern：

https://docs.mongodb.com/mast...

writeConcern：

https://docs.mongodb.com/mast...