如何使用 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();

还有 mongoose 交易文档 有点误导，或者至少它可能更多描述性的.它在示例中所指的 db 实际上是 Mongoose Connection 实例，而不是底层的 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<在模块化代码中获取它/code> 属性， 连接建立后的任何时间.这通常在服务器路由处理程序等内部是安全的，因为在调用代码时将有一个数据库连接.

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.

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

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屋！