首先

来源：https://docs.mongodb.com/manual/core/transactions-in-applications/
出典：https://docs.mongodb.com/manual/core/transactions-in-applications/

只有一个节点不被支持。如果执行将会出错。

Exception in thread "main" com.mongodb.MongoClientException: Sessions are not supported by the MongoDB cluster to which this client is connected
    at com.mongodb.client.internal.MongoClientImpl.startSession(MongoClientImpl.java:127)
    at com.mongodb.client.internal.MongoClientImpl.startSession(MongoClientImpl.java:113)
    at mongodb.MongoDBTest.main(MongoDBTest.java:24)

准备复制集环境。

创建各个节点的文件夹节点1: /data/mongo-replicaset/节点1
节点2: /data/mongo-replicaset/节点2
节点3: /data/mongo-replicaset/节点3

启动每个节点

mongod --replSet my-set --dbpath /data/mongo-replicaset/node1 --logpath /data/mongo-replicaset/node1/node1.log --port 27001
mongod --replSet my-set --dbpath /data/mongo-replicaset/node2 --logpath /data/mongo-replicaset/node2/node2.log --port 27002
mongod --replSet my-set --dbpath /data/mongo-replicaset/node3 --logpath /data/mongo-replicaset/node3/node3.log --port 27003

初始化 Relica Set连接到Node1并进行初始化。

rs.initiate(
{
        "_id" : "my-set",
        "members" : [
                {
                        "_id" : 0,
                        "host" : "localhost:27001"
                },
                {
                        "_id" : 1,
                        "host" : "localhost:27002"
                },
                {
                        "_id" : 2,
                        "host" : "localhost:27003"
                }
        ]
});

用rs.status()来检查状态

好的。准备工作已经完成。

尝试使用JAVA进行事务处理

向不同的数据库添加多条记录。

package mongodb;

import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.TransactionOptions;
import com.mongodb.WriteConcern;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.TransactionBody;

import java.util.Date;

import org.bson.Document;

public class MongoDBTest {

    public static void main(String[] args) {
        // MongoDBクライントを生成する
        MongoClient client = MongoClients.create("mongodb://localhost:27001,localhost:27002,localhost:27003");

        // コレクションが存在しないと、トランザクションの処理は実行されないため、コードでコレクションを作成する。
        //（一回だけ。手動でコレクションを追加してもよい）
        client.getDatabase("front").getCollection("access_log").withWriteConcern(WriteConcern.MAJORITY)
                .insertOne(new Document("dummy", 0));
        client.getDatabase("server").getCollection("users").withWriteConcern(WriteConcern.MAJORITY)
                .insertOne(new Document("dummy", 0));

        // セッションを生成する
        ClientSession session = client.startSession();

        // Options定義
        TransactionOptions txnOptions = TransactionOptions.builder().readPreference(ReadPreference.primary())
                .readConcern(ReadConcern.LOCAL).writeConcern(WriteConcern.MAJORITY).build();

        TransactionBody txnBody = new TransactionBody<String>() {
            public String execute() {
                MongoCollection<Document> frontAccessLog = client.getDatabase("front").getCollection("access_log");
                MongoCollection<Document> serverUsers = client.getDatabase("server").getCollection("users");

                // アクセスログ
                Document accessLog = new Document();
                accessLog.append("log", "xxxx");
                accessLog.append("acccessDate", new Date());

                frontAccessLog.insertOne(session, accessLog);

                // ユーザーデータ
                Document user = new Document();
                user.append("lastName", "tanaka");
                user.append("firstName", "tarou");
                user.append("createDate", new Date());

                serverUsers.insertOne(session, user);

                return "OK";
            }
        };

        try {
            // 同じトランザクションで処理
            session.withTransaction(txnBody, txnOptions);

        } catch (RuntimeException e) {
            // 異常処理

        } finally {
            session.close();
        }

        // クライントを閉じる
        client.close();
    }
}

在同一个数据库中处理多条记录

package mongodb;

import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.TransactionOptions;
import com.mongodb.WriteConcern;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.TransactionBody;
import com.mongodb.client.model.Filters;

import java.util.Date;

import org.bson.Document;

public class MongoDBTest {

    public static void main(String[] args) {
        // MongoDBクライントを生成する
        MongoClient client = MongoClients.create("mongodb://localhost:27001,localhost:27002,localhost:27003");

        // セッションを生成する
        ClientSession session = client.startSession();

        // Options定義
        TransactionOptions txnOptions = TransactionOptions.builder().readPreference(ReadPreference.primary())
                .readConcern(ReadConcern.LOCAL).writeConcern(WriteConcern.MAJORITY).build();

        TransactionBody txnBody = new TransactionBody<String>() {
            public String execute() {
                MongoCollection<Document> frontAccessLog = client.getDatabase("front").getCollection("access_log");

                // ダミーデータを削除
                frontAccessLog.deleteOne(Filters.eq("dummy", 0));

                // アクセスログ
                Document accessLog = new Document();
                accessLog.append("log", "yyyyy");
                accessLog.append("acccessDate", new Date());

                frontAccessLog.insertOne(session, accessLog);

                // 異常にする
                throw new RuntimeException("処理失敗です。");
                // return "OK";
            }
        };

        try {
            // 同じトランザクションで処理
            session.withTransaction(txnBody, txnOptions);

        } catch (RuntimeException e) {
            // 異常処理
            e.printStackTrace();
        } finally {
            session.close();
        }

        // クライントを閉じる
        client.close();
    }
}

你特意慢动作，可以确认DB数据没有变化。
当删除throw new RuntimeException(“処理失敗です。”);后，成功地进行了预期的数据删除和插入。

事务URL：https://docs.mongodb.com/manual/core/transactions/
复制：https://docs.mongodb.com/manual/replication/
搭建副本集：https://docs.mongodb.com/manual/tutorial/deploy-replica-set/
副本集认证设置：https://docs.mongodb.com/manual/tutorial/deploy-replica-set-with-keyfile-access-control/

以上

#技术文章