如何解决 MongoDB 中缺少事务的问题？

Question

我知道这里也有类似的问题，但如果我需要事务或使用atomic operations 或two-phase commit，它们要么是telling me 以切换回常规RDBMS 系统。第二种解决方案似乎是最佳选择。第三个我不想遵循，因为似乎很多事情都可能出错，而且我无法在各个方面都对其进行测试。我很难重构我的项目以执行原子操作。我不知道这是否来自我有限的观点（我到目前为止只使用过 SQL 数据库），还是实际上无法做到。

我们想在我们公司对 MongoDB 进行试点测试。我们选择了一个比较简单的项目——短信网关。它允许我们的软件向蜂窝网络发送 SMS 消息，而网关则完成了这项繁琐的工作：实际上是通过不同的通信协议与提供商进行通信。网关还管理消息的计费。每个申请该服务的客户都必须购买一些积分。系统在发送消息时自动减少用户的余额，如果余额不足则拒绝访问。此外，由于我们是第三方 SMS 提供商的客户，我们也可能与他们有自己的余额。我们还必须跟踪这些。

我开始考虑如果我降低一些复杂性（外部计费、排队的 SMS 发送），如何使用 MongoDB 存储所需的数据。来自 SQL 世界，我将为用户创建一个单独的表，另一个用于 SMS 消息，以及一个用于存储有关用户余额的交易。假设我为 MongoDB 中的所有集合创建了单独的集合。

在这个简化的系统中想象一个具有以下步骤的 SMS 发送任务：

1. 检查用户是否有足够的余额；如果信用不足，则拒绝访问

2. 发送消息并将其存储在 SMS 集合中，其中包含详细信息和费用（在实时系统中，消息将具有 status 属性，并且任务将提取它以进行传递并设置 SMS 的价格根据当前状态）

3. 通过发送消息的成本减少用户的余额

4. 在事务集合中记录事务

现在有什么问题？ MongoDB 只能对一个文档进行原子更新。在前面的流程中，可能会出现某种错误，消息被存储在数据库中，但用户的余额没有更新和/或交易没有记录。

我想出了两个想法：

• 为用户创建单个集合，并将余额作为字段存储，用户相关的交易和消息作为子文档存储在用户的文档中。因为我们可以原子地更新文档，这实际上解决了事务问题。缺点：如果用户发送很多短信，文档会变大，可能会达到4MB文档的限制。也许我可以在这种情况下创建历史文档，但我认为这不是一个好主意。另外，如果我将越来越多的数据推送到同一个大文档中，我也不知道系统会多快。

• 为用户创建一个集合，为事务创建一个。可以有两种交易：信用购买，余额为正数，消息发送，余额为负数。交易可能有一个子文件；例如，在 messages sent 中，SMS 的详细信息可以嵌入到交易中。缺点：我不存储当前的用户余额，所以每次用户尝试发送消息时我都必须计算它来判断消息是否可以通过。恐怕随着存储交易数量的增加，这种计算会变慢。

我对选择哪种方法有点困惑。还有其他解决方案吗？我在网上找不到任何关于如何解决这类问题的最佳实践。我想很多试图熟悉 NoSQL 世界的程序员一开始都面临着类似的问题。

Answer 1

从 4.0 开始，MongoDB 将具有多文档 ACID 事务。该计划是首先启用副本集部署中的那些，然后是分片集群。 MongoDB 中的事务感觉就像开发人员熟悉的关系数据库中的事务一样——它们将是多语句，具有相似的语义和语法（如 start_transaction 和 commit_transaction）。重要的是，启用事务的 MongoDB 更改不会影响不需要它们的工作负载的性能。

更多详情请见here。

拥有分布式事务并不意味着您应该像在表格关系数据库中那样对数据进行建模。拥抱文档模型的力量并遵循数据建模的良好和推荐practices。

Answer 2

检查this，由 Tokutek 提供。他们为 Mongo 开发了一个插件，它不仅承诺交易，还承诺提高性能。

Answer 3

直截了当地说：如果事务完整性是必须，那么不要使用 MongoDB，而只使用系统中支持事务的组件。为了为非 ACID 兼容的组件提供与 ACID 类似的功能，在组件之上构建一些东西是非常困难的。根据个别用例，以某种方式将操作分为事务性和非事务性操作可能是有意义的......

Answer 4

现在有什么问题？ MongoDB 只能对一个文档进行原子更新。在前面的流程中，可能会发生某种错误，消息被存储在数据库中，但用户的余额没有减少和/或交易没有被记录。

这不是一个真正的问题。您提到的错误是逻辑（错误）或 IO 错误（网络、磁盘故障）。这种错误可能会使无事务存储和事务存储处于不一致状态。例如，如果它已经发送了短信，但是在存储信息时发生了错误 - 它不能回滚短信发送，这意味着它不会被记录，用户余额不会减少等等。

这里真正的问题是用户可以利用竞争条件并发送比他的余额允许的更多的消息。这也适用于 RDBMS，除非您使用余额字段锁定在事务内部发送 SMS（这将是一个很大的瓶颈）。作为 MongoDB 的一个可能的解决方案是首先使用findAndModify 来减少余额并检查它，如果它是负数，则不允许发送并退还金额（原子增量）。如果是肯定的，请继续发送，如果失败，请退还金额。还可以维护余额历史集合以帮助修复/验证余额字段。

Answer 5

项目很简单，但你必须支持交易支付，这使得整个事情变得困难。因此，例如，具有数百个集合（论坛、聊天、广告等）的复杂门户系统在某些方面更简单，因为如果您丢失了论坛或聊天条目，没有人会真正关心。另一方面，如果您丢失了一笔付款交易，这是一个严重的问题。

所以，如果你真的想要一个 MongoDB 的试点项目，请选择一个在 方面简单的项目。

Answer 6

由于正当的原因，MongoDB 中没有事务。这是使 MongoDB 更快的原因之一。

在你的情况下，如果交易是必须的，mongo 似乎不太合适。

可能是 RDMBS + MongoDB，但这会增加复杂性，并使管理和支持应用程序变得更加困难。

Answer 7

这可能是我找到的关于为 mongodb 实现类似事务的功能的最佳博客。！

同步标志：最适合从主文档复制数据

Job Queue：非常通用，解决了 95% 的情况。无论如何，大多数系统都需要至少有一个作业队列！

两阶段提交：此技术确保每个实体始终拥有达到一致状态所需的所有信息

Log Reconciliation：最强大的技术，非常适合金融系统

版本控制：提供隔离并支持复杂结构

阅读本文了解更多信息：https://dzone.com/articles/how-implement-robust-and

Answer 8

这已经晚了，但认为这对将来会有帮助。我使用Redis 制作queue 来解决这个问题。

• 要求：
  下图显示 2 个动作需要同时执行，但动作 1 的第 2 阶段和第 3 阶段需要在动作 2 的第 2 阶段开始之前完成或相反（一个阶段可以是请求 REST api、数据库请求或执行 javascript 代码......）。

• 队列如何帮助您
  队列确保lock()和release()之间的每个块代码在许多函数中不会同时运行，使它们隔离。

  function action1() {
    phase1();
    queue.lock("action_domain");
    phase2();
    phase3();
    queue.release("action_domain");
  }
  
  function action2() {
    phase1();
    queue.lock("action_domain");
    phase2();
    queue.release("action_domain");
  }
  

• 如何建立队列
  我将只关注在后端站点上构建队列时如何避免race conditon 部分。如果你不知道队列的基本概念，来here。
  下面的代码只展示了概念，你需要以正确的方式实现。

  function lock() {
    if(isRunning()) {
      addIsolateCodeToQueue(); //use callback, delegate, function pointer... depend on your language
    } else {
      setStateToRunning();
      pickOneAndExecute();
    }
  }
  
  function release() {
    setStateToRelease();
    pickOneAndExecute();
  }
  

但你需要isRunning() setStateToRelease() setStateToRunning() 隔离它自己，否则你将再次面临竞争条件。为此，我选择 Redis 用于 ACID 目的和可扩展性。
Redisdocument说说它的事务：

事务中的所有命令都被序列化并执行 依次。永远不会发生另一个人发出的请求 客户端在 Redis 执行过程中提供服务 交易。这保证了命令作为 单个隔离操作。

P/s：
我使用 Redis 是因为我的服务已经在使用它，您可以使用任何其他支持隔离的方式来做到这一点。
我的代码中的 action_domain 上面是当您只需要用户 A 阻止用户 A 的操作 2 的操作 1 时，不要阻止其他用户。这个想法是为每个用户锁定一个唯一的密钥。

Answer 9

事务现在在 MongoDB 4.0 中可用。示例here

// Runs the txnFunc and retries if TransientTransactionError encountered

function runTransactionWithRetry(txnFunc, session) {
    while (true) {
        try {
            txnFunc(session);  // performs transaction
            break;
        } catch (error) {
            // If transient error, retry the whole transaction
            if ( error.hasOwnProperty("errorLabels") && error.errorLabels.includes("TransientTransactionError")  ) {
                print("TransientTransactionError, retrying transaction ...");
                continue;
            } else {
                throw error;
            }
        }
    }
}

// Retries commit if UnknownTransactionCommitResult encountered

function commitWithRetry(session) {
    while (true) {
        try {
            session.commitTransaction(); // Uses write concern set at transaction start.
            print("Transaction committed.");
            break;
        } catch (error) {
            // Can retry commit
            if (error.hasOwnProperty("errorLabels") && error.errorLabels.includes("UnknownTransactionCommitResult") ) {
                print("UnknownTransactionCommitResult, retrying commit operation ...");
                continue;
            } else {
                print("Error during commit ...");
                throw error;
            }
       }
    }
}

// Updates two collections in a transactions

function updateEmployeeInfo(session) {
    employeesCollection = session.getDatabase("hr").employees;
    eventsCollection = session.getDatabase("reporting").events;

    session.startTransaction( { readConcern: { level: "snapshot" }, writeConcern: { w: "majority" } } );

    try{
        employeesCollection.updateOne( { employee: 3 }, { $set: { status: "Inactive" } } );
        eventsCollection.insertOne( { employee: 3, status: { new: "Inactive", old: "Active" } } );
    } catch (error) {
        print("Caught exception during transaction, aborting.");
        session.abortTransaction();
        throw error;
    }

    commitWithRetry(session);
}

// Start a session.
session = db.getMongo().startSession( { mode: "primary" } );

try{
   runTransactionWithRetry(updateEmployeeInfo, session);
} catch (error) {
   // Do something with error
} finally {
   session.endSession();
}