事务

本页内容

概述

事务 API
回调 API
核心 API
其他信息
API 文档

概述

在本指南中，您可以了解如何使用MongoDB C++驱动程序来执行事务。事务允许您运行一系列操作，直到事务提交之前，这些操作都不会更改任何数据。如果事务中的任何操作返回错误，驱动程序将取消事务并丢弃所有更改，使其不会变得可见。

在MongoDB中，事务在逻辑会话中运行。会话是一组相关的读或写操作，您打算按顺序运行。会话使一组操作在符合ACID规范的事务中实现因果一致性，即满足原子性、一致性、隔离性和持久性期望的事务。MongoDB保证涉及事务操作的数据保持一致性，即使操作遇到意外错误。

使用C++驱动程序时，您可以从mongocxx::client实例创建一个新的会话。然后，您可以使用生成的mongocxx::client_session实例执行事务。我们建议您重复使用客户端进行多个会话和事务，而不是每次都实例化一个新的客户端。

警告

仅使用创建它的mongocxx::client与mongocxx::client_session一起使用。使用与不同client的client_session会导致操作错误。

重要

mongocxx::client的实例不是线程安全的。每个mongoxcc::client实例及其子实例，包括mongocxx::client_session，应同时由单个线程使用。有关更多信息，请参阅线程和Fork安全性指南。

事务API

MongoDB C++驱动程序提供了一种回调API和核心API来管理事务的生命周期。在开始事务之前，您必须调用start_session()方法来实例化一个mongocxx::client_session。然后，您可以使用以下任一API执行事务

回调API：高级API，用于管理事务的生命周期并自动集成错误处理逻辑。
核心API：低级API，允许您管理事务的生命周期并实现自定义错误处理逻辑。

提示

要了解有关错误处理的更多信息，请参阅MongoDB服务器手册中的事务错误处理部分。

回调API

使用回调API允许MongoDB C++驱动程序管理您的交易生命周期。要实现此API，请在您的mongocxx::client_session上调用with_transaction()方法，并传入一个回调函数，指定您想要运行的操作序列。with_transaction()方法启动事务，执行回调函数，并在遇到错误时提交您的交易或结束事务。如果您的交易遇到TransientTransactionError或UnknownTransactionCommitResult错误，则with_transaction()方法将重新运行事务。

以下代码使用回调API在sample_mflix数据库的movies和comments集合中执行事务。此代码执行以下操作

使用start_session()方法从客户端启动会话。
定义一个回调函数，指定在事务期间要执行的操作。
创建一个选项对象，准备设置事务操作的写关注。要了解更多有关读取和写入语义的信息，请参阅MongoDB服务器手册中的读取关注/写关注/读取优先级部分。
调用with_transaction()方法来管理事务，将回调函数和选项对象作为参数传递。

// Establish a connection to the MongoDB deployment 
mongocxx::instance instance{};
mongocxx::client client(mongocxx::uri{"<connectionString>"});
// Define database and collection variables
auto db = client["sample_mflix"];
auto movies_collection = db["movies"];
auto comments_collection = db["comments"]; 
// Define a callback specifying the sequence of operations to perform during the transaction
mongocxx::client_session::with_transaction_cb callback = [&](mongocxx::client_session* session) {
    // Important::  You must pass the session to the operations.
    movies_collection.insert_one(*session, make_document(kvp("title", "Parasite")));
    comments_collection.insert_one(*session, make_document(kvp("name", "Anjali Patel"), 
                                                           kvp("text", "This is my new favorite movie!")));
};
// Define an options instance to explicitly set the write concern for the transaction operations  
mongocxx::options::transaction opts;
mongocxx::write_concern wc;
wc.acknowledge_level(mongocxx::write_concern::level::k_majority);
opts.write_concern(wc);
// Start a client session
auto session = client.start_session();
try {
    // Start a transaction, execute the operations in the callback function, and commit the results 
    session.with_transaction(callback, opts);
} catch (const mongocxx::exception& e) {
    std::cout << "An exception occurred: " << e.what() << std::endl;
    return EXIT_FAILURE;
}
return EXIT_SUCCESS;

核心API

使用核心API来管理事务的生命周期。为了实现此API，您必须显式调用 mongocxx::client_session 接口中的方法以启动事务、提交活动事务以及在发生错误时结束事务。核心API不会自动包含错误处理逻辑，而是允许您实现自定义的错误处理逻辑，包括 TransientTransactionError 和 UnknownTransactionCommitResult。

以下表格描述了由 mongocxx::client_session 接口提供的核心API方法

方法	描述
`start_transaction()`	在当前客户端会话上启动一个新的事务。接受一个可选的 `mongocxx::options::transaction` 实例作为参数以设置选项。有关选项的完整列表，请参阅API文档中的 mongocxx::options::transaction。如果选项配置不正确、存在网络或其他短暂故障，或者存在其他错误（如已存在事务的会话），则引发异常。如果返回了带有 `TransientTransactionError` 标签的错误，您可以结束事务，然后带期望它成功的期望重新尝试。有关此方法的更多信息，请参阅MongoDB服务器手册中的startTransaction()指南。
`commit_transaction()`	提交当前客户端会话上的活动事务。如果选项配置错误、发生网络或其他短暂故障，或者存在其他错误，例如没有进行事务的会话，则会引发异常。如果返回了带有 `UnknownTransactionCommitResult` 标签的错误，您可以结束事务然后尝试重新提交，预期在提交的事务满足设置的写关注时将成功。要了解更多关于此方法的信息，请参阅 MongoDB 服务器手册中的 commitTransaction() 指南。
`abort_transaction()`	结束当前客户端会话上的活动事务。如果选项配置错误或存在其他错误，例如没有进行事务的会话，则会引发异常。要了解更多关于此方法的信息，请参阅 MongoDB 服务器手册中的 abortTransaction() 指南。

提示

mongocxx::client_session 类还提供了检索和修改会话属性的方法。要了解更多信息，请参阅 API 文档中的 mongocxx::client_session。

以下代码使用核心 API 执行事务，将文档插入到 sample_mflix 数据库中的 movies 和 comments 集合。此代码执行以下操作

使用start_session()方法从客户端启动会话。
创建一个选项对象，准备设置事务操作的写关注。要了解更多有关读取和写入语义的信息，请参阅MongoDB服务器手册中的读取关注/写关注/读取优先级部分。
调用 start_transaction() 方法开始事务，并将选项对象作为参数传入。
运行操作，将文档插入到 sample_mflix 数据库中的集合，并将活动会话传递给每个操作。如果操作遇到错误，则整个事务将被中止。如果错误具有 TransientTransactionError 标签，则重试事务。
使用 commit_transaction() 方法提交活动事务。如果提交遇到带有 UnknownTransactionCommitResult 标签的错误，则重试提交。

// Establish a connection to the MongoDB deployment 
mongocxx::instance instance{};
mongocxx::client client(mongocxx::uri{"<connectionString>"});
// Runs the txn_func and retries if TransientTransactionError occurs
using transaction_func = std::function<void(mongocxx::client_session& session)>;
auto run_with_retry = [](transaction_func txn_func, 
                         mongocxx::client_session& session) {
    while (true) {
        try {
            txn_func(session);  // performs transaction.
            break;
        } catch (const mongocxx::operation_exception& oe) {
            std::cout << "Transaction aborted. Caught exception during transaction." 
                      << std::endl;
            // If transient error, retry the whole transaction.
            if (oe.has_error_label("TransientTransactionError")) {
                std::cout << "TransientTransactionError, retrying transaction..."
                          << std::endl;
                continue;
            } else {
                throw oe;
            }
        }
    }
};
// Commits the active transaction and retries commit if UnknownTransactionCommitResult occurs
auto commit_with_retry = [](mongocxx::client_session& session) {
    while (true) {
        try {
            session.commit_transaction();  // Uses write concern set at transaction start.
            std::cout << "Transaction committed." 
                      << std::endl;
            break;
        } catch (const mongocxx::operation_exception& oe) {
            // Can retry commit
            if (oe.has_error_label("UnknownTransactionCommitResult")) {
                std::cout << "UnknownTransactionCommitResult, retrying commit..."
                          << std::endl;
                continue;
            } else {
                std::cout << "Error during commit..." 
                          << std::endl;
                throw oe;
            }
        }
    }
    
};
auto txn_func = [&](mongocxx::client_session& session) {
    auto& client = session.client(); 
    // Define database and collection variables
    auto db = client["sample_mflix"];
    auto movies_collection = db["movies"];
    auto comments_collection = db["comments"];
    // Define an options instance to explicitly set the write concern for the transaction operations  
    mongocxx::options::transaction opts;
    mongocxx::write_concern wc;
    wc.acknowledge_level(mongocxx::write_concern::level::k_majority);
    opts.write_concern(wc);
    session.start_transaction(opts);
    // Attempt to insert documents into database collections
    try {
        movies_collection.insert_one(session, make_document(kvp("title", "Parasite")));
        comments_collection.insert_one(session, make_document(kvp("name", "Anjali Patel"), 
                                                              kvp("text", "This is my new favorite movie!")));
    } catch (const mongocxx::operation_exception& oe) {
        std::cout << "Caught exception during transaction, aborting." 
                  << std::endl;
        session.abort_transaction();
        throw oe;
    }
    commit_with_retry(session);
};
// Start a client session
auto session = client.start_session();
try {
    run_with_retry(txn_func, session);
} catch (const mongocxx::operation_exception& oe) {
    // Do something with error 
    throw oe;
}