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

Question

我知道这里也有类似的问题，但它们要么告诉我如果我需要事务，就切换回常规 RDBMS 系统或使用 原子操作 或 两阶段提交。第二种解决方案似乎是最好的选择。第三个我不想遵循，因为似乎很多事情都可能出错，而且我无法在每个方面都进行测试。我很难重构我的项目来执行原子操作。我不知道这是否来自我有限的观点（到目前为止我只使用过 SQL 数据库），或者它是否实际上无法完成。

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

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

想象一下，在这个简化的系统中，短信发送任务包含以下步骤：

1. 检查用户是否有足够的余额；如果没有足够的信用

2. 发送消息并将其存储在 SMS 集合中，并附上详细信息和费用（在实时系统中，消息将具有 status 属性，并且任务会拾取它用于发送并根据其当前状态设置 SMS 的价格）

3. 根据发送消息的费用减少用户的余额

4. 在交易集合中记录交易

   p>

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

我提出了两个想法：

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

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

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

Answer 1

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

有关更多详细信息，请参阅此处。

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

Answer 2

检查这个，作者：Tokutek 。他们为 Mongo 开发了一个插件，不仅可以实现交易，还可以提高性能。

Answer 3

言归正传：如果事务完整性是必须，那么就不要使用 MongoDB，而只使用系统中支持事务的组件。在组件之上构建一些东西以便为不符合 ACID 的组件提供类似 ACID 的功能是极其困难的。根据各个用例，以某种方式将操作分为事务性操作和非事务性操作可能是有意义的......

Answer 4

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

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

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

Answer 5

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

因此，如果您确实想要一个 MongoDB 试点项目，请选择一个在这方面简单的项目。

Answer 6

出于正当原因，MongoDB 中不存在事务。这是让 MongoDB 更快的因素之一。

就您而言，如果必须进行交易，那么 mongo 似乎不适合。

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

Answer 7

这可能是我发现的关于为 mongodb 实现事务类功能的最好的博客。！

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

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

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

日志协调：最强大的技术，非常适合金融系统

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

阅读此内容以获取更多信息：https://dzone.com/articles/how-implement-robust-and

Answer 8

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

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

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

  <块引用>

  函数action1() {
    阶段1();
    队列.lock(“action_domain”);
    阶段2();
    阶段3();
    队列.release(“action_domain”);
  }
  
  函数动作2() {
    阶段1();
    队列.lock(“action_domain”);
    阶段2();
    队列.release(“action_domain”);
  }
  

• 如何构建队列
  我将只关注在后端站点上构建队列时如何避免竞争条件部分。如果您不了解队列的基本概念，请访问此处。
  下面的代码仅展示了概念，您需要以正确的方式实现。

  <块引用>

  函数锁() {
    如果（正在运行（））{
      添加隔离代码到队列（）； //使用回调、委托、函数指针...取决于您的语言
    } 别的 {
      setStateToRunning();
      pickOneAndExecute();
    }
  }
  
  函数释放（）{
    setStateToRelease();
    pickOneAndExecute();
  }
  

但是您需要 isRunning() setStateToRelease() setStateToRunning() 隔离它的自身，否则您将再次面临竞争条件。为此，我选择 Redis 来实现 ACID 目的且可扩展。
Redis 文档谈论它的事务：

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

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

Answer 9

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

// 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();
}