一文读懂 | 驱动程序 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 transaction
session.startTransaction( { readConcern: { level: "local" }, writeConcern: { w: "majority" } } );
// Operations inside the transaction
try {
   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 start
session.commitTransaction();
session.endSession();