一文读懂 驱动程序 API

// WithTransactionExample is an example of using the Session.WithTransaction function.func WithTransactionExample() {  ctx := context.Background()  // For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.  // uri := "mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl"  // For a sharded cluster, connect to the mongos instances; e.g.  // uri := "mongodb://mongos0.example.com:27017,mongos1.example.com:27017/"  var uri string
  clientOpts := options.Client().ApplyURI(uri)  client, err := mongo.Connect(ctx, clientOpts)  if err != nil {    panic(err)  }  defer func() { _ = client.Disconnect(ctx) }()
  // Prereq: Create collections.  wcMajority := writeconcern.New(writeconcern.WMajority(), writeconcern.WTimeout(1*time.Second))  wcMajorityCollectionOpts := options.Collection().SetWriteConcern(wcMajority)  fooColl := client.Database("mydb1").Collection("foo", wcMajorityCollectionOpts)  barColl := client.Database("mydb1").Collection("bar", wcMajorityCollectionOpts)
  // Step 1: Define the callback that specifies the sequence of operations to perform inside the transaction.  callback := func(sessCtx mongo.SessionContext) (interface{}, error) {    // Important: You must pass sessCtx as the Context parameter to the operations for them to be executed in the    // transaction.    if _, err := fooColl.InsertOne(sessCtx, bson.D{{"abc", 1}}); err != nil {      return nil, err    }    if _, err := barColl.InsertOne(sessCtx, bson.D{{"xyz", 999}}); err != nil {      return nil, err    }
    return nil, nil  }
  // Step 2: Start a session and run the callback using WithTransaction.  session, err := client.StartSession()  if err != nil {    panic(err)  }  defer session.EndSession(ctx)
  result, err := session.WithTransaction(ctx, callback)  if err != nil {    panic(err)  }  fmt.Printf("result: %v\n", result)}

runTransactionWithRetry := func(sctx mongo.SessionContext, txnFn func(mongo.SessionContext) error) error {    for {      err := txnFn(sctx) // Performs transaction.      if err == nil {        return nil      }
      log.Println("Transaction aborted. Caught exception during transaction.")
      // If transient error, retry the whole transaction      if cmdErr, ok := err.(mongo.CommandError); ok && cmdErr.HasErrorLabel("TransientTransactionError") {        log.Println("TransientTransactionError, retrying transaction...")        continue      }      return err    }  }
  commitWithRetry := func(sctx mongo.SessionContext) error {    for {      err := sctx.CommitTransaction(sctx)      switch e := err.(type) {      case nil:        log.Println("Transaction committed.")        return nil      case mongo.CommandError:        // Can retry commit        if e.HasErrorLabel("UnknownTransactionCommitResult") {          log.Println("UnknownTransactionCommitResult, retrying commit operation...")          continue        }        log.Println("Error during commit...")        return e      default:        log.Println("Error during commit...")        return e      }    }  }
  // Updates two collections in a transaction.  updateEmployeeInfo := func(sctx mongo.SessionContext) error {    employees := client.Database("hr").Collection("employees")    events := client.Database("reporting").Collection("events")
    err := sctx.StartTransaction(options.Transaction().      SetReadConcern(readconcern.Snapshot()).      SetWriteConcern(writeconcern.New(writeconcern.WMajority())),    )    if err != nil {      return err    }
    _, err = employees.UpdateOne(sctx, bson.D{{"employee", 3}}, bson.D{{"$set", bson.D{{"status", "Inactive"}}}})    if err != nil {      sctx.AbortTransaction(sctx)      log.Println("caught exception during transaction, aborting.")      return err    }    _, err = events.InsertOne(sctx, bson.D{{"employee", 3}, {"status", bson.D{{"new", "Inactive"}, {"old", "Active"}}}})    if err != nil {      sctx.AbortTransaction(sctx)      log.Println("caught exception during transaction, aborting.")      return err    }
    return commitWithRetry(sctx)  }
  return client.UseSessionWithOptions(    ctx, options.Session().SetDefaultReadPreference(readpref.Primary()),    func(sctx mongo.SessionContext) error {      return runTransactionWithRetry(sctx, updateEmployeeInfo)    },  )}

// Create collections:db.getSiblingDB("mydb1").foo.insert( {abc: 0}, { writeConcern: { w: "majority", wtimeout: 2000 } } );db.getSiblingDB("mydb2").bar.insert( {xyz: 0}, { writeConcern: { w: "majority", wtimeout: 2000 } } );
// Start a session.session = db.getMongo().startSession( { readPreference: { mode: "primary" } } );
coll1 = session.getDatabase("mydb1").foo;coll2 = session.getDatabase("mydb2").bar;
// Start a transactionsession.startTransaction( { readConcern: { level: "local" }, writeConcern: { w: "majority" } } );
// Operations inside the transactiontry {   coll1.insertOne( { abc: 1 } );   coll2.insertOne( { xyz: 999 } );} catch (error) {   // Abort transaction on error   session.abortTransaction();   throw error;}
// Commit the transaction using write concern set at transaction startsession.commitTransaction();
session.endSession();