事务

Overview

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

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

注意

C++驱动程序不支持在单个事务中运行并行操作。

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

警告

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

重要

mongocxx::client 的实例不是线程安全的。每个 mongoxcc::client实例及其子实例（包括 mongocxx::client_session）一次应由一个线程使用。要学习；了解更多信息，请参阅线程和分叉安全指南。

因果一致性(Causal Consistency)

MongoDB在某些客户端会话中实现因果一致性。因果一致性模型ACID 一致性保证在分布式系统中，会话中的操作按因果顺序运行。客户端观察到的结果与因果关系或操作之间的依赖关系一致。示例，如果您执行一系列操作，其中一个操作在逻辑上依赖于另一个操作的结果，则任何后续读取都会反映这种依赖关系。

为了保证因果一致性，客户端端会话必须满足以下要求：

启动会话时，驱动程序必须启用因果一致性选项。该选项默认启用。
操作必须在单个线程的单个会话中运行。否则，会话或线程必须相互传达optime和集群时间值。要查看传达这些值的两个会话的示例，请参阅MongoDB Server手册中的因果一致性示例。
您必须使用 mongocxx::read_concern::level::k_majority读关注（read concern）。
您必须使用 mongocxx::write_concern::level::k_majority写关注（write concern）。这是默认的写关注（write concern）值。

下表描述了因果一致会话提供的ACID 一致性保证：

保证	说明
读取写入操作	读取操作会反映之前写入操作的结果。
单调读取	读取操作不会返回反映比先前读取操作更早的数据状态的结果。
单调写入	如果写入操作必须先于其他写入操作，则服务器会先运行此写入操作。示例，如果调用 `insert_one()` 插入文档，然后调用 `update_one()` 修改插入的文档，则服务器首先运行插入操作。
读取后写入	如果写入操作必须在其他读取操作之后执行，服务器会先执行读取操作。示例，如果您调用 `find_one()` 来检索文档，然后调用 `delete_one()` 来删除检索到的文档，则服务器会先运行查找操作。

提示

要学习；了解有关本节中提到的概念的更多信息，请参阅以下MongoDB Server手册条目：

事务 API

MongoDB C++驱动程序提供了 Callback API和 Core API来管理ACID 事务方式。在开始ACID 事务之前，必须调用 start_session() 方法来实例化 mongocxx::client_session。然后，您可以使用以下任一 API 来执行ACID 事务：

Callback API ：管理ACID 事务生命周期并自动包含错误处理逻辑的高级API 。
核心API ：低级API ，允许您管理ACID 事务的生命周期并实现自定义错误处理逻辑。

提示

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

回调 API

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

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

使用 start_session() 方法从客户端启动会话。
定义回调函数，指定ACID 事务期间要执行的操作。
创建一个选项对象以准备为ACID 事务操作设立写关注（write concern）。要学习；了解有关读取和写入语义的更多信息，请参阅MongoDB Server手册中的读关注/写关注/读取偏好部分。
调用 with_transaction() 方法来管理ACID 事务，并将回调函数和选项对象作为参数传递。

// 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;

Core API

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

下表描述了 mongocxx::client_session 接口提供的 Core API方法：

方法	说明
`start_transaction()`	Starts a new transaction on the current client session. Accepts an optional `mongocxx::options::transaction` instance as an argument to set options. For a full list of options, see mongocxx::options::transaction in the API documentation. Raises an exception if the options are misconfigured, if there are network or other transient failures, or if there are other errors such as a session with a transaction already in progress. If an error is returned with the `TransientTransactionError` label, you can end the transaction and then retry it with the expectation that it will succeed. To learn more about this method, see the startTransaction() guide in the MongoDB Server manual.
`commit_transaction()`	Commits the active transaction on the current client session. Raises an exception if options are misconfigured, if there are network or other transient failures, or if there are other errors such as a session with no transaction in progress. If an error is returned with the `UnknownTransactionCommitResult` label, you can end the transaction and then retry it with the expectation that it will succeed when the committed transaction satisfies the set write concern. To learn more about this method, see the commitTransaction() guide in the MongoDB Server manual.
`abort_transaction()`	Ends the active transaction on the current client session. Raises an exception if the options are misconfigured or if there are other errors such as a session with no transaction in progress. To learn more about this method, see the abortTransaction() guide in the MongoDB Server manual.

提示

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

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

使用 start_session() 方法从客户端启动会话。
创建一个选项对象以准备为ACID 事务操作设立写关注（write concern）。要学习；了解有关读取和写入语义的更多信息，请参阅MongoDB Server手册中的读关注/写关注/读取偏好部分。
调用 start_transaction() 方法启动ACID 事务，并将选项对象作为参数传入。
运行操作以将文档插入到 sample_mflix数据库的集合中，并将活动会话传递给每个操作。如果操作遇到错误，整个ACID 事务将中止。如果错误具有标签 TransientTransactionError，则会重试ACID 事务。
使用 commit_transaction() 方法提交ACID 事务。如果提交遇到标签为 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;
}

Overview

注意

警告

重要

因果一致性(Causal Consistency)

提示

事务 API

提示

回调 API

Core API

提示

更多信息

API 文档