mongodb4.0多文档事务尝鲜

brew install mongodb
brew services start mongodb

vim /usr/local/etc/mongod.conf
......
# 这里添加复制集名称的选项
replication:
  replSetName: rs0
  
brew services start mongodb

mongo
> db.adminCommand( { getParameter: 1, featureCompatibilityVersion: 1 } )
> // 如果特性兼容版本小于4.0，则要设置为4.0
> db.adminCommand( { setFeatureCompatibilityVersion: "4.0" } )
> exit

mongo
> rs.initiate()
> rs.status()
> exit

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

mongo
> use hr
> db.createCollection("employees");
> use reporting
> db.createCollection("events")
> exit

mongo
> load("test.js")
Transaction committed.
true
> exit

package main

import (
	"context"
	"fmt"
	"github.com/mongodb/mongo-go-driver/mongo"
	"github.com/mongodb/mongo-go-driver/core/command"
	"github.com/mongodb/mongo-go-driver/bson"
)

func main() {
	client, err := mongo.Connect(context.Background(), "mongodb://localhost:27017", nil)
	panicIfErr(err)

	sess, err := client.StartSession()
	panicIfErr(err)
	defer sess.EndSession(context.Background())

	runTransactionWithRetry(updateEmployeeInfo, client, sess)
}

// Runs the txnFunc and retries if TransientTransactionError encountered

func runTransactionWithRetry(txnFunc func(*mongo.Client, *mongo.Session) error, client *mongo.Client, sess *mongo.Session) error {
	var err error
	for
	{
		err = txnFunc(client, sess) // performs transaction
		if err != nil {
			if v, ok := err.(command.Error); ok {
				if contains(v.Labels, command.TransientTransactionError) {
					fmt.Println("TransientTransactionError, retrying transaction ...")
					continue
				}
			}
			return err
		}
		break
	}
	return nil
}



// Retries commit if UnknownTransactionCommitResult encountered

func commitWithRetry(sess *mongo.Session) error {
	for
	{
		err := sess.CommitTransaction(context.Background()) // Uses write concern set at transaction start.
		if err != nil {
			if v, ok := err.(command.Error); ok {
				if contains(v.Labels, command.UnknownTransactionCommitResult) {
					fmt.Println("UnknownTransactionCommitResult, retrying commit operation ...")
					continue
				}
			}
			fmt.Println("Error during commit ...")
			return err
		}
		fmt.Println("Transaction committed.")
		break
	}
	return nil
}

// Updates two collections in a transactions

func updateEmployeeInfo(client *mongo.Client, sess *mongo.Session) error{
	employeesCollection := client.Database("hr").Collection("employees")
	eventsCollection := client.Database("reporting").Collection("events")
	err := sess.StartTransaction()
	if err != nil {
		return err
	}
	_, err = employeesCollection.UpdateOne(context.Background(), bson.NewDocument(
		bson.EC.Int32("employee", 3),
	),bson.NewDocument(
		bson.EC.SubDocumentFromElements("$set",
			bson.EC.String("status", "Inactive"),
		),
	), sess)
	if err != nil {
		fmt.Println("Caught exception during transaction, aborting.")
		sess.AbortTransaction(context.Background())
		return err
	}
	_, err = eventsCollection.InsertOne(context.Background(), bson.NewDocument(
		bson.EC.Int32("employee", 3),
		bson.EC.SubDocumentFromElements("status",
			bson.EC.String("new", "Inactive"),
			bson.EC.String("old", "Active"),
		),
	), sess)
	if err != nil {
		fmt.Println("Caught exception during transaction, aborting.")
		sess.AbortTransaction(context.Background())
		return err
	}
	return commitWithRetry(sess)
}

func panicIfErr(err error) {
	if err != nil {
		panic(err)
	}
}

func contains(slice []string, item string) bool {
	set := make(map[string]struct{}, len(slice))
	for _, s := range slice {
		set[s] = struct{}{}
	}

	_, ok := set[item]
	return ok
}