如何在 Meteor 方法调用中实现事务

Question

假设我有 2 个集合“PlanSubscriptions”和“ClientActivations”。我正在对这两个集合进行连续插入。

后一个依赖前一个，如果任何一个事务失败，那么整个操作必须回滚。

如何在 Meteor 1.4 中实现这一点？

Answer 1

在 Meteor 中没有办法做到这一点，因为 mongodb 不是 ACID 兼容的数据库。它具有单文档更新原子性，但不是多文档更新原子性，这就是您使用两个集合的情况。 来自 mongo 文档：

单个写操作修改多个文档时，每个文档的修改都是原子的，但是整个操作不是原子的，其他操作可能交错。

有一种方法可以隔离多文档更新的可见性，但这可能不是您所需要的。

使用 $isolated 运算符，一旦写入操作修改了第一个文档，影响多个文档的写入操作可以防止其他进程交错。这确保了在写入操作完成或出错之前没有客户端看到更改。 独立的写入操作不提供“全有或全无”的原子性。也就是说，写入操作期间的错误不会回滚错误之前的所有更改。

但是，有几个库试图在应用程序级别解决问题。我建议看看fawn

在您的情况下，您恰好有两个依赖集合，可以利用两阶段提交技术。在此处阅读更多信息：two-phase-commits

Answer 2

由于 MongoDB 不支持原子性，您必须使用方法链来管理它。

您可以编写一个方法，例如transaction，您将在其中调用PlanSubscriptions.insert(data, callback)。然后在回调函数中，如果第一次插入成功，您将调用ClientActivations.insert(data, callback1)，如果第二次插入成功，则在callback1 中返回真值，否则返回假值。如果第一次插入返回错误，则无需执行任何操作，但如果第二次插入返回错误，则删除从第一个集合中插入的 id。

我可以建议以下结构：

'transaction'(){
    PlanSubscriptions.insert(data, (error, result)=>{
        if(result){
            // result contains the _id
            let id_plan = result;
            ClientActivations.insert(data, (error, result)=>{
                if(result){
                    // result contains the _id
                    return true;
                }
                else if(error){
                    PlanSubscriptions.remove(id_plan);
                    return false;
                }
            })
        }
        else if(error){
            return false;
        }
    })
}

Answer 3

嗯，我自己想通了。

我添加了一个包babrahams:transactions

在服务器端 Meteor Method 调用中，我调用了 tx 包全局公开的对象。整个服务器端Meteor.method({}) 如下所示。

import { Meteor } from 'meteor/meteor';
import {PlanSubscriptions} from '/imports/api/plansubscriptions/plansubscriptions.js';
import {ClientActivations} from '/imports/api/clientactivation/clientactivations.js';

Meteor.methods({
    'createClientSubscription' (subscriptionData, clientActivationData) {
            var txid;
            try {
            txid = tx.start("Adding Subscription to our database");

            PlanSubscriptions.insert(subscriptionData, {tx: true})

            ClientActivations.insert(activation, {tx: true});

            tx.commit();

            return true;
        } catch(e){
            tx.undo(txid);
        }
        return false;
    }
});

每次插入我都添加了{tx : true}，这表明它是事务的一部分。

服务器控制台输出：

I20170523-18:43:23.544(5.5)? Started "Adding Subscription to our database" with
transaction_id: vdJQvFgtyZuWcinyF
I20170523-18:43:23.547(5.5)? Pushed insert command to stack: vdJQvFgtyZuWcinyF
I20170523-18:43:23.549(5.5)? Pushed insert command to stack: vdJQvFgtyZuWcinyF
I20170523-18:43:23.551(5.5)? Beginning commit with transaction_id: vdJQvFgtyZuWcinyF
I20170523-18:43:23.655(5.5)? Executed insert
I20170523-18:43:23.666(5.5)? Executed insert
I20170523-18:43:23.698(5.5)? Commit reset transaction manager to clean state

欲了解更多信息，您可以转到链接：https://github.com/JackAdams/meteor-transactions

注意：我使用的是 Meteor 1.4.4.2

Answer 4

仅将此链接分享给未来的读者：

https://forums.meteor.com/t/solved-transactions-with-mongodb-meteor-methods/48677

import { MongoInternals } from 'meteor/mongo';

// utility async function to wrap async raw mongo operations with a transaction
const runTransactionAsync = async asyncRawMongoOperations => {

    // setup a transaction
    const { client } = MongoInternals.defaultRemoteCollectionDriver().mongo;
    const session = await client.startSession();
    await session.startTransaction();
    try {
        // running the async operations
        let result = await asyncRawMongoOperations(session);
        await session.commitTransaction();
        // transaction committed - return value to the client
        return result;
    } catch (err) {
        await session.abortTransaction();
        console.error(err.message);
        // transaction aborted - report error to the client
        throw new Meteor.Error('Database Transaction Failed', err.message);
    } finally {
        session.endSession();
    }
};

import { runTransactionAsync } from '/imports/utils'; // or where you defined it

Meteor.methods({
    async doSomething(arg) {
        // remember to check method input first

        // define the operations we want to run in transaction
        const asyncRawMongoOperations = async session => {

            // it's critical to receive the session parameter here
            // and pass it to every raw operation as shown below
            const item = await collection1.rawCollection().findOne(arg, { session: session });
            const response = await collection2.rawCollection().insertOne(item, { session: session });

            // if Mongo or you throw an error here runTransactionAsync(..) will catch it
            // and wrap it with a Meteor.Error(..) so it will arrive to the client safely

            return 'whatever you want'; // will be the result in the client
        };

        let result = await runTransactionAsync(asyncRawMongoOperations);
        return result;
    }
});