在 Node.js 中使用 MongoDB 事务

MongoDB事务

事务介绍

在 MongoDB 中，对单个文档的操做是原子的。因为您可使用嵌入的文档和数组来捕获单个文档结构中的数据之间的关系，而不是跨多个文档和集合进行规范化，所以这种单一文档的原子性消除了对多文档的需求许多实际用例的事务。

对于须要对多个文档（在单个或多个集合中）进行读取和写入原子化的状况，MongoDB 支持多文档事务。对于分布式事务，事务可用于多个操做、集合、数据库、文档和分片。

事务和原子性

分布式事务和多单据事务 从 MongoDB 4.2 开始，这两个术语是同义词。分布式事务是指分片群集和副本集上的多文档交易记录。多文档事务（不管是在分片群集仍是副本集上）也称为从 MongoDB 4.2 开始的分布式事务。 对于须要对多个文档（在单个或多个集合中）进行读取和写入原子化的状况，MongoDB 支持多文档事务：

在版本 4.0中，MongoDB 支持副本集上的多文档事务。

在版本 4.2中，MongoDB 引入了分布式事务，这增长了对分片群集上的多文档事务的支持，并合并了对副本集上多文档事务的现有支持。

要在 MongoDB 4.2 部署（副本集和分片群集）上使用事务，客户端必须使用为 MongoDB 4.2 更新的 MongoDB 驱动程序。

多文档事务是原子的（即提供"全无"命题）：

当事务提交时，事务中所作的全部数据更改都将保存在事务外部并可见。也就是说，事务不会提交其某些更改，而回滚其余更改。

在事务提交以前，事务中所作的数据更改在事务外部不可见。

可是，当事务写入多个分片时，并不是全部外部读取操做都须要等待提交的事务的结果在分片中可见。例如，若是提交事务，写入 1 在分片 A 上可见，但在分片 B 上还没有显示写入 2，则读取时的外部读取"local"能够读取写入 1 的结果，而看不到写入 2。

当事务停止时，事务中所作的全部数据更改将被丢弃，而不会变得可见。例如，若是事务中的任何操做失败，事务将停止，而且事务中所作的全部数据更改将被丢弃，而不会变得可见。

准备工做

MongoDB 使用事务的前提是 MongoDB 版本大于 4.0,须要配置 MongoDB 工做模式为副本集,单个 MongoDB 节点不足支持事务,由于 MongoDB 事务至少须要两个节点。其中一个是主节点，负责处理客户端请求，其他的都是从节点，负责复制主节点上的数据。mongodb各个节点常见的搭配方式为：一主一从、一主多从。主节点记录在其上的全部操做oplog，从节点按期轮询主节点获取这些操做，而后对本身的数据副本执行这些操做，从而保证从节点的数据与主节点一致。

部署	功能 兼容性版本
副本集	4.0
分片集群	4.2

命令行部署

启动实例

mongod --replSet rs --dbpath=磁盘目录 --port=27017
mongod --replSet rs --dbpath=磁盘目录 --port=37017
复制代码

Mongo shell

$mongo --port=27017

MongoDB shell version v4.2.3
connecting to: mongodb://127.0.0.1:27017/?compressors=disabled&gssapiServiceName=mongodb
Implicit session: session { "id" : UUID("b0a2609c-6aa1-466a-849f-ba0e9f5e3d3a") }
MongoDB server version: 4.2.3
...
复制代码

副本集配置

var config={
     _id:"rs",
     members:[
         {_id:0,host:"127.0.0.1:27017"},
         {_id:1,host:"127.0.0.1:37017"},
]}；
rs.initiate(config)
// 成功后会返回相似以下信息
{
    "ok" : 1,
    "operationTime" : Timestamp(1522810920, 1),
    "$clusterTime" : {
        "clusterTime" : Timestamp(1522810920, 1),
        "signature" : {
            "hash" : BinData(0,"AAAAAAAAAAAAAAAAAAAAAAAAAAA="),
            "keyId" : NumberLong(0)
        }
    }
}

复制代码

容器

Docker 部署

//指定 MongoDB 版本 > 4.0,也能够指定latest 
docker pull mongo:4.2.3
复制代码

启动 Docker 容器

docker run --name m0 -p 37017:27017 -d mongo:4.2.3 --replSet "rs"
docker run --name m0 -p 47017:27017 -d mongo:4.2.3 --replSet "rs"
docker run --name m0 -p 57017:27017 -d mongo:4.2.3 --replSet "rs"
复制代码

mongo shell

// 先进入 Docker 容器交互模式
docker exec -it CONTAINERID /bin/bash
剩余配置方法与命令行部署相同
复制代码

Node.js 中使用 MongoDB事务

使用MongoDB驱动

// 对于副本集来讲 Uri 中须要包含副本集名称,和成员 URI
  // const uri = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
  // 对于分片集群，链接到mongo集群实例
  // const uri = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'
  
  const client = new MongoClient(uri);
  await client.connect();
  
  await client
    .db('demo')
    .collection('cats')
    .insertOne({ name: 0 }, { w: 'majority' });

  await client
    .db('demo')
    .collection('cats')
    .insertOne({ name: 0 }, { w: 'majority' });

  // 第一步 启动 session,事务的全部操做都基于 session
  const session = client.startSession();

  // 第二步 定义事务选项
  const transactionOptions = {
    readPreference: 'primary',
    readConcern: { level: 'local' },
    writeConcern: { w: 'majority' }
  };

  // 第三步：使用withTransaction启动事务、执行回调和提交
  
  try {
    await session.withTransaction(async () => {
      const coll1 = client.db('demo').collection('cats');
      const coll2 = client.db('demo').collection('cats');

      // 必须将会话传递给操做

      await coll1.insertOne({ name: 1 }, { session });
      await coll2.insertOne({ name: 999 }, { session });
    }, transactionOptions);
  } finally {
    await session.endSession();
    await client.close();
  }

复制代码

Egg.js 框架中使用 mongoose 执行事务

注意事项 参考了(博客园博主兜兜的文章)

• 需使用mongoose.connection对集合进行事务操做，其余model的CRUD部分方法不支持事务

mongoose.connection.collection('集合名') // 注：集合名须要小写且加s,如model为Cat,集合名这里应写为cats

复制代码

• 触发Schema定义的中间件默认值须要构造model实例

const CatSchema = new Schema({
    name: {
        type: String
        default: 'cat'
    },
    created: {
     type: Date,
     default: Date.now
  }
})
 
const Cat = mongoose.model('Cat', CatSchema)
 
new Cat() // 触发中间件
复制代码

• insertOne，findOneAndUpdate等方法对数据的新增，需上面第二点进行依赖，不然直接insertOne 插入一条数据，定义的默认值不会触发，如created字段,chema内部定义的type: Schema.ObjectId的相应字段，insertOne插入后都会变成字符串类型，不是Schema.ObjectId类型

// 解决方式
//新增
 
const Cat= new Cat();
const data = {name: 5}
for (let key in data) {
      Cat[key] = data[key];
    }
db.collection('cats').insertOne(Cat);
 
// 查询修改
 
db.collection('cats')
.findOneAndUpdate({_id: mongoose.Types.ObjectId(你的id)}, {$set: {name: 修改值}})
复制代码

• 副本集模式下,集合不会被自动建立须要手动建立集合才能进行操做

/*
2020-03-14 14:05:16,525 ERROR 2476 [-/127.0.0.1/-/30ms GET /] nodejs.Error: MongoError: Cannot create namespace demo.cats in multi-document transaction.
*/
复制代码

扩展 context

// /app/extend/context.js
module.exports = {
  async getSession(opt = {
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' },
  }) {
    const { mongoose } = this.app;
    const session = await mongoose.startSession(opt);
    await session.startTransaction();
    return session;
  },
};

复制代码

模型

'use strict';
module.exports = app => {
  const CatSchema = new app.mongoose.Schema({
    name: {
      type: String,
      default: 'cat',
    },
    pass: {
      type: String,
      default: 'cat',
    },
    created: {
      type: Date,
      default: Date.now,
    },
  });

  const Cat = app.mongoose.model('Cat', CatSchema);

  new Cat(); // 触发中间件
  return Cat;
};

复制代码

执行事务

const { mongoose } = this.ctx.app;
    const session = await this.ctx.getSession();
    const db = mongoose.connection;
    try {
      let data = { name : 'ceshi' };
      const Cat = new this.ctx.model.Cat();
      for (let key in data) {
        Cat[key] = data[key]
      }
      await db
        .collection('cats')
        .insertOne(Cat, { session });
      // 提交事务
      await session.commitTransaction();
      return 'ok';
    } catch (err) {
      // 回滚事务
      const res = await session.abortTransaction();
      this.ctx.logger.error(new Error(err));
    } finally {
      await session.endSession();
    }
    // 执行后,数据库中多了一条 { name: 'ceshi'} 的记录
复制代码

事务回滚

const { mongoose } = this.ctx.app;
    const session = await this.ctx.getSession();
    const db = mongoose.connection;
    try {
      let data = { name : 'ceshi' };
      const Cat = new this.ctx.model.Cat();
      for (let key in data) {
        Cat[key] = data[key]
      }
      await db
        .collection('cats')
        .insertOne(Cat, { session });
      await this.ctx.model.Cat.deleteMany({ name: 'ceshi' }, { session });
      // 手动抛出异常
      await this.ctx.throw();
      // 提交事务
      await session.commitTransaction();
      return 'ok';
    } catch (err) {
      // 回滚事务
      await session.abortTransaction();
      this.ctx.logger.error(new Error(err));
    } finally {
      // 结束事务
      await session.endSession();
    }
    // 手动抛出异常后,事务回滚,查看数据库能够看到,插入和删除文档都没有生效
    /*
    2020-03-14 14:01:38,503 ERROR 2476 [-/127.0.0.1/-/25ms GET /] nodejs.Error: InternalServerError: Internal Server Error
    */
复制代码