如何使用Mongoose使用MongoDB事务? [英] How to use MongoDB transaction using Mongoose?

问题描述

我正在使用MongoDB Atlas云( https://cloud.mongodb.com/)和Mongoose库.

I am using MongoDB Atlas cloud(https://cloud.mongodb.com/) and Mongoose library.

我尝试使用事务处理概念创建多个文档，但是它不起作用. 我没有任何错误.但是，回滚似乎无法正常工作.

I tried to create multiple documents using transaction concept, but it is not working. I am not getting any error. but, it seems rollback is not working properly.

app.js

//*** more code here

var app = express();

require('./models/db');

//*** more code here

models/db.js

var mongoose = require( 'mongoose' );

// Build the connection string
var dbURI = 'mongodb+srv://mydb:pass@cluster0-****.mongodb.net/mydb?retryWrites=true';

// Create the database connection
mongoose.connect(dbURI, {
  useCreateIndex: true,
  useNewUrlParser: true,
});

// Get Mongoose to use the global promise library
mongoose.Promise = global.Promise;

models/user.js

const mongoose = require("mongoose");

const UserSchema = new mongoose.Schema({
  userName: {
    type: String,
    required: true
  },
  pass: {
    type: String,
    select: false
  }
});

module.exports = mongoose.model("User", UserSchema, "user");

myroute.js

const db = require("mongoose");
const User = require("./models/user");

router.post("/addusers", async (req, res, next) => {

    const SESSION = await db.startSession();

    await SESSION.startTransaction();

    try {

          const newUser = new User({
            //*** data for user ***
          });
          await newUser.save();

          //*** for test purpose, trigger some error ***
          throw new Error("some error");

          await SESSION.commitTransaction();

          //*** return data 

    } catch (error) {
            await SESSION.abortTransaction();
    } finally {
            SESSION.endSession();
    }    

 });

以上代码可正常运行，但仍会在数据库中创建用户.它假定要回滚创建的用户，并且集合应该为空.

Above code works without error, but it still creates user in the DB. It suppose to rollback the created user and the collection should be empty.

我不知道我在这里错过了什么.有人可以让我知道这是怎么回事吗?

I don't know what I have missed here. Can anyone please let me know whats wrong here?

应用，模型，架构和路由器位于不同的文件中.

app, models, schema and router are in different files.

推荐答案

对于在事务期间处于活动状态的所有读/写操作，您需要在选项中包括session.只有这样，它们才真正应用于您可以回滚的事务范围.

You need to include the session within the options for all read/write operations which are active during a transaction. Only then are they actually applied to the transaction scope where you are able to roll them back.

列表更为完整，仅使用更经典的Order/OrderItems建模，对于具有一定关系交易经验的大多数人来说，它们应该非常熟悉:

As a bit more complete listing, and just using the more classic Order/OrderItems modelling which should be pretty familiar to most people with some relational transactions experience:

const { Schema } = mongoose = require('mongoose');

// URI including the name of the replicaSet connecting to
const uri = 'mongodb://localhost:27017/trandemo?replicaSet=fresh';
const opts = { useNewUrlParser: true };

// sensible defaults
mongoose.Promise = global.Promise;
mongoose.set('debug', true);
mongoose.set('useFindAndModify', false);
mongoose.set('useCreateIndex', true);

// schema defs

const orderSchema = new Schema({
  name: String
});

const orderItemsSchema = new Schema({
  order: { type: Schema.Types.ObjectId, ref: 'Order' },
  itemName: String,
  price: Number
});

const Order = mongoose.model('Order', orderSchema);
const OrderItems = mongoose.model('OrderItems', orderItemsSchema);

// log helper

const log = data => console.log(JSON.stringify(data, undefined, 2));

// main

(async function() {

  try {

    const conn = await mongoose.connect(uri, opts);

    // clean models
    await Promise.all(
      Object.entries(conn.models).map(([k,m]) => m.deleteMany())
    )

    let session = await conn.startSession();
    session.startTransaction();

    // Collections must exist in transactions
    await Promise.all(
      Object.entries(conn.models).map(([k,m]) => m.createCollection())
    );

    let [order, other] = await Order.insertMany([
      { name: 'Bill' },
      { name: 'Ted' }
    ], { session });

    let fred = new Order({ name: 'Fred' });
    await fred.save({ session });

    let items = await OrderItems.insertMany(
      [
        { order: order._id, itemName: 'Cheese', price: 1 },
        { order: order._id, itemName: 'Bread', price: 2 },
        { order: order._id, itemName: 'Milk', price: 3 }
      ],
      { session }
    );

    // update an item
    let result1 = await OrderItems.updateOne(
      { order: order._id, itemName: 'Milk' },
      { $inc: { price: 1 } },
      { session }
    );
    log(result1);

    // commit
    await session.commitTransaction();

    // start another
    session.startTransaction();

    // Update and abort
    let result2 = await OrderItems.findOneAndUpdate(
      { order: order._id, itemName: 'Milk' },
      { $inc: { price: 1 } },
      { 'new': true, session }
    );
    log(result2);

    await session.abortTransaction();

    /*
     * $lookup join - expect Milk to be price: 4
     *
     */

    let joined = await Order.aggregate([
      { '$match': { _id: order._id } },
      { '$lookup': {
        'from': OrderItems.collection.name,
        'foreignField': 'order',
        'localField': '_id',
        'as': 'orderitems'
      }}
    ]);
    log(joined);


  } catch(e) {
    console.error(e)
  } finally {
    mongoose.disconnect()
  }

})()

因此，我通常建议使用小写形式调用变量session，因为这是选项"对象的键名，在所有操作中都需要它.将其保持为小写约定还允许使用诸如ES6对象分配之类的东西:

So I would generally recommend calling the variable session in lowercase, since this is the name of the key for the "options" object where it is required on all operations. Keeping this in the lowercase convention allows for using things like the ES6 Object assignment as well:

const conn = await mongoose.connect(uri, opts);

...

let session = await conn.startSession();
session.startTransaction();

猫鼬关于交易的文档有点误导，或者至少可能会更多描述性的.在示例中所谓的db实际上是猫鼬连接实例，并且而不是底层的Db甚至mongoose全局导入，因为有些人可能会误解这一点.请注意，在清单和上面的摘录中，这是从mongoose.connect()获取的，应该作为可从共享导入中访问的内容保存在代码中.

Also the mongoose documentation on transactions is a little misleading, or at least it could be more descriptive. What it refers to as db in the examples is actually the Mongoose Connection instance, and not the underlying Db or even the mongoose global import as some may misinterpret this. Note in the listing and above excerpt this is obtained from mongoose.connect() and should be kept within your code as something you can access from a shared import.

或者，您甚至可以通过 mongoose.connection 以模块化代码的形式进行抓取.属性，在建立连接后的任何时间 .在服务器路由处理程序之类的东西内部这通常是安全的，因为在调用代码时将建立数据库连接.

Alternately you can even grab this in modular code via the mongoose.connection property, at any time after a connection has been established. This is usually safe inside things such as server route handlers and the like since there will be a database connection by the time that code is called.

代码还演示了不同模型方法中session的用法:

The code also demonstrates the session usage in the different model methods:

let [order, other] = await Order.insertMany([
  { name: 'Bill' },
  { name: 'Ted' }
], { session });

let fred = new Order({ name: 'Fred' });
await fred.save({ session });

所有基于find()的方法以及基于update()或insert()和delete()的方法都具有最终的选项块"，该会话键和值应在此位置. save()方法的唯一参数是此选项块.这就是告诉MongoDB在引用的会话上将这些操作应用于当前事务的原因.

All the find() based methods and the update() or insert() and delete() based methods all have a final "options block" where this session key and value are expected. The save() method's only argument is this options block. This is what tells MongoDB to apply these actions to the current transaction on that referenced session.

以几乎相同的方式，在提交事务之前，任何未指定session选项的find()或类似请求都不会在进行该事务时看到数据状态.事务完成后，修改后的数据状态仅对其他操作可用.请注意，这会影响文档中所述.

In much the same way, before a transaction is committed any requests for a find() or similar which do not specify that session option do not see the state of the data whilst that transaction is in progress. The modified data state is only available to other operations once the transaction completes. Note this has effects on writes as covered in the documentation.

发出中止"时:

// Update and abort
let result2 = await OrderItems.findOneAndUpdate(
  { order: order._id, itemName: 'Milk' },
  { $inc: { price: 1 } },
  { 'new': true, session }
);
log(result2);

await session.abortTransaction();

对活动事务的任何操作都将从状态中删除，并且不应用.因此，它们对于随后的结果操作是不可见的.在此处的示例中，文档中的值将递增，并且将在当前会话中显示检索到的值5.但是，在session.abortTransaction()之后，将恢复文档的先前状态.请注意，任何未在同一会话上读取数据的全局上下文，除非已提交，否则不会看到状态更改.

Any operations on the active transaction are removed from state and are not applied. As such they are not visible to resulting operations afterwards. In the example here the value in the document is incremented and will show a retrieved value of 5 on the current session. However after session.abortTransaction() the previous state of the document is reverted. Note that any global context which was not reading data on the same session, does not see that state change unless committed.

这应该给出总体概述.可以添加更多的复杂性来处理各种级别的写入失败和重试，但是文档和许多示例中已经对此进行了广泛介绍，或者可以回答更具体的问题.

That should give the general overview. There is more complexity that can be added to handle varying levels of write failure and retries, but that is already extensively covered in documentation and many samples, or can be answered to a more specific question.

作为参考，包含的清单的输出如下所示:

For reference, the output of the included listing is shown here:

Mongoose: orders.deleteMany({}, {})
Mongoose: orderitems.deleteMany({}, {})
Mongoose: orders.insertMany([ { _id: 5bf775986c7c1a61d12137dd, name: 'Bill', __v: 0 }, { _id: 5bf775986c7c1a61d12137de, name: 'Ted', __v: 0 } ], { session: ClientSession("80f827fe077044c8b6c0547b34605cb2") })
Mongoose: orders.insertOne({ _id: ObjectId("5bf775986c7c1a61d12137df"), name: 'Fred', __v: 0 }, { session: ClientSession("80f827fe077044c8b6c0547b34605cb2") })
Mongoose: orderitems.insertMany([ { _id: 5bf775986c7c1a61d12137e0, order: 5bf775986c7c1a61d12137dd, itemName: 'Cheese', price: 1, __v: 0 }, { _id: 5bf775986c7c1a61d12137e1, order: 5bf775986c7c1a61d12137dd, itemName: 'Bread', price: 2, __v: 0 }, { _id: 5bf775986c7c1a61d12137e2, order: 5bf775986c7c1a61d12137dd, itemName: 'Milk', price: 3, __v: 0 } ], { session: ClientSession("80f827fe077044c8b6c0547b34605cb2") })
Mongoose: orderitems.updateOne({ order: ObjectId("5bf775986c7c1a61d12137dd"), itemName: 'Milk' }, { '$inc': { price: 1 } }, { session: ClientSession("80f827fe077044c8b6c0547b34605cb2") })
{
  "n": 1,
  "nModified": 1,
  "opTime": {
    "ts": "6626894672394452998",
    "t": 139
  },
  "electionId": "7fffffff000000000000008b",
  "ok": 1,
  "operationTime": "6626894672394452998",
  "$clusterTime": {
    "clusterTime": "6626894672394452998",
    "signature": {
      "hash": "AAAAAAAAAAAAAAAAAAAAAAAAAAA=",
      "keyId": 0
    }
  }
}
Mongoose: orderitems.findOneAndUpdate({ order: ObjectId("5bf775986c7c1a61d12137dd"), itemName: 'Milk' }, { '$inc': { price: 1 } }, { session: ClientSession("80f827fe077044c8b6c0547b34605cb2"), upsert: false, remove: false, projection: {}, returnOriginal: false })
{
  "_id": "5bf775986c7c1a61d12137e2",
  "order": "5bf775986c7c1a61d12137dd",
  "itemName": "Milk",
  "price": 5,
  "__v": 0
}
Mongoose: orders.aggregate([ { '$match': { _id: 5bf775986c7c1a61d12137dd } }, { '$lookup': { from: 'orderitems', foreignField: 'order', localField: '_id', as: 'orderitems' } } ], {})
[
  {
    "_id": "5bf775986c7c1a61d12137dd",
    "name": "Bill",
    "__v": 0,
    "orderitems": [
      {
        "_id": "5bf775986c7c1a61d12137e0",
        "order": "5bf775986c7c1a61d12137dd",
        "itemName": "Cheese",
        "price": 1,
        "__v": 0
      },
      {
        "_id": "5bf775986c7c1a61d12137e1",
        "order": "5bf775986c7c1a61d12137dd",
        "itemName": "Bread",
        "price": 2,
        "__v": 0
      },
      {
        "_id": "5bf775986c7c1a61d12137e2",
        "order": "5bf775986c7c1a61d12137dd",
        "itemName": "Milk",
        "price": 4,
        "__v": 0
      }
    ]
  }
]

这篇关于如何使用Mongoose使用MongoDB事务?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！