bulkwrite не поддерживает многодокументную транзакцию с использованием mongo-go-driver - PullRequest
Новая
       3

bulkwrite не поддерживает многодокументную транзакцию с использованием mongo-go-driver

0 голосов
/ 04 декабря 2018

Я использую mongo-go-driver 0.0.18 для создания массовой записи, которая состоит из «NewUpdateManyModel» и нескольких «NewInsertOneModel».Мой сервер Монго - атлас М10 с наборами реплик.Я построил некоторые процедуры, чтобы проверить, являются ли транзакции атомарными, результат показывает, что каждая массовая запись не атомарна, они будут мешать друг другу.Мне интересно, поддерживает ли mongo-go-driver транзакции с несколькими документами? 

func insertUpdateQuery(counter int, col *mongo.Collection, group *sync.WaitGroup){
var operations []mongo.WriteModel
var items = []item{}
items=append(items,item{"Name":strconv.Itoa(counter),"Description":"latest one"})

for _,v := range items{
    operations = append(operations, mongo.NewInsertOneModel().Document(v))
}

updateOperation := mongo.NewUpdateManyModel()
updateOperation.Filter(bson.D{
    {"Name", bson.D{
        {"$ne", strconv.Itoa(counter)},
    }},
})
updateOperation.Update(bson.D{
    {"$set", bson.D{
        {"Description", strconv.Itoa(counter)},
    }},
},)
operations = append(operations,updateOperation)

bulkOps:=options.BulkWrite()
result, err := col.BulkWrite(
    context.Background(),
    operations,
    bulkOps,
)
if err != nil{
    fmt.Println("err:",err)
}else{
    fmt.Printf("IU: %+v \n",result)
}

group.Done()

}


func retrieveQuery(group *sync.WaitGroup, col *mongo.Collection){
var results []item
qctx:=context.Background()
qctx, c := context.WithTimeout(qctx, 10*time.Second)
defer c()
cur, err := col.Find(qctx, nil)
if err != nil {
    log.Fatal(err)
}
defer cur.Close(context.Background())
res := item{}
for cur.Next(context.Background()) {
    err := cur.Decode(&res)
    if err != nil {
        log.Println(err)
    }else {
        results=append(results,res)
    }
}
if err := cur.Err(); err != nil {
    log.Println(err)
}

fmt.Println("res:",results)

group.Done()
}



func main()  {
ctx := context.Background()
ctx, cancel := context.WithTimeout(ctx,10*time.Second)
defer cancel()

uri := "..."
client, err := mongo.NewClient(uri)
if err != nil {
    fmt.Printf("todo: couldn't connect to mongo: %v", err)
}
defer cancel()
err = client.Connect(ctx)
if err != nil {
    fmt.Printf("todo: mongo client couldn't connect with background context: %v", err)
}
col:=client.Database("jistest").Collection("Rules")

wg :=&sync.WaitGroup{}

for i:=0; i<100; i++{
    wg.Add(2)
    go insertUpdateQuery(i,col,wg)
    go retrieveQuery(wg,col)

}
wg.Wait()
fmt.Println("All Done!")
}

1 Ответ

0 голосов
/ 10 декабря 2018

Мне интересно, поддерживает ли mongo-go-driver транзакцию с несколькими документами?

mongo-go-driver поддерживает транзакцию с несколькими документами, поскольку v0.0.12 (в настоящее время в бета-версии 0.1.0).

Многодокументарные транзакции MongoDB связаны с сеансом.То есть вы запускаете транзакцию для сеанса.При использовании любого из официально поддерживаемых драйверов MongoDB вы должны передавать сеанс каждой операции в транзакции.

Ваш пример не использует ни сеанс, ни транзакции.Пример многодокументной транзакции в mongo-go-driver (v0.1.0) приведен ниже: 

client, err := mongo.NewClient("<MONGODB URI>")
if err != nil { return err }
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
err = client.Connect(ctx)
if err != nil { return err }

session, err := client.StartSession()
database := client.Database("databaseName")
collection := database.Collection("collectionName")

err = mongo.WithSession(ctx, session, func(sctx mongo.SessionContext) error {
        // Start a transaction in the session 
        sctx.StartTransaction()

        var operations []mongo.WriteModel
        // Create an insert one operation
        operations = append(operations,
                            mongo.NewInsertOneModel().Document(
                            bson.D{{"Name", counter}, 
                                   {"Description", "latest"}}))

        // Create an update many operation
        updateOperation := mongo.NewUpdateManyModel()
        updateOperation.Filter(bson.D{{"Name", bson.D{
            {"$ne", counter},
        }}})
        updateOperation.Update(bson.D{
            {"$set", bson.D{
                {"Description", counter},
            }},
        })
        operations = append(operations, updateOperation)

        // Execute bulkWrite operation in a transactional session.
        _, err := collection.BulkWrite(sctx, operations)
        if err != nil {
            fmt.Println(err)
            return err
        }
        // Committing transaction
        session.CommitTransaction(sctx)
        return nil
})
session.EndSession(ctx)

См. Также Транзакции и Retryable Writes для примеров повторных транзакций.

Я создал несколько процедур, чтобы проверить, являются ли транзакции атомарными

Просто помните о том, как выполняются процессы.Например, в зависимости от условий гонки, вы можете получить последнюю производительность, перезаписав результат.то есть 

transaction 1 finished 
transaction 2 finished 
transaction 3 and transaction 4 conflict 
transaction 5 finished
...
...