So implementieren Sie eine verteilte Transaktion über Mysql, Redis und Mongo

Der folgende Code ist in Golang. Andere Sprachen wie C#, PHP, Java finden Sie hier:DTM SDKs

func SagaAdjustBalance(db dtmcli.DB, uid int, amount int) error {
    _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?" , amount, uid)
    return err
}

Dieser Code führt hauptsächlich die Anpassung des Guthabens des Benutzers in der Datenbank durch. In unserem Beispiel wird dieser Teil des Codes nicht nur für die Vorwärtsoperation von Saga verwendet, sondern auch für die Kompensationsoperation, bei der nur ein negativer Betrag zur Kompensation übergeben werden muss.

Für Redis und Mongo wird der Geschäftscode ähnlich gehandhabt, es werden lediglich die entsprechenden Salden erhöht oder verringert.

So stellen Sie Idempotenz sicher

Wenn beim Saga-Transaktionsmuster ein vorübergehender Fehler im Subtransaktionsdienst auftritt, wird die fehlgeschlagene Operation wiederholt. Dieser Fehler kann vor oder nach dem Commit der Subtransaktion auftreten, daher muss die Subtransaktionsoperation idempotent sein.

DTM stellt Hilfstabellen und Hilfsfunktionen bereit, um Benutzern zu helfen, Idempotenz schnell zu erreichen. Für Mysql wird eine Hilfstabelle barrier erstellt Wenn der Benutzer in der Geschäftsdatenbank eine Transaktion zur Anpassung des Kontostands startet, wird zunächst Gid eingefügt in der barrier Tisch. Wenn eine doppelte Zeile vorhanden ist, schlägt die Einfügung fehl und die Ausgleichsanpassung wird übersprungen, um die Idempotenz sicherzustellen. Der Code, der die Hilfsfunktion verwendet, lautet wie folgt:

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error {
        return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult)
    })
}))

Mongo handhabt Idempotenz ähnlich wie Mysql, daher gehe ich nicht noch einmal ins Detail.

Redis handhabt Idempotenz anders als Mysql, hauptsächlich wegen des Unterschieds im Transaktionsprinzip. Redis-Transaktionen werden hauptsächlich durch die atomare Ausführung von Lua sichergestellt. Die DTM-Hilfsfunktion passt die Balance über ein Lua-Skript an. Vor dem Anpassen des Kontostands wird Gid abgefragt in Redis. Wenn Gid vorhanden ist, wird die Balance-Anpassung übersprungen; wenn nicht, wird Gid aufgezeichnet und führen Sie die Balance-Einstellung durch. Der für die Hilfsfunktion verwendete Code lautet wie folgt:

app.POST(BusiAPI+"/SagaRedisTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), -reqFrom(c).Amount, 7*86400)
}))

Wie man eine Entschädigung durchführt

Für Saga müssen wir uns auch mit dem Kompensationsvorgang befassen, aber die Kompensation ist nicht einfach eine umgekehrte Anpassung, und es gibt viele Fallstricke, die Sie kennen sollten.

Einerseits muss bei der Kompensation Idempotenz berücksichtigt werden, da die im vorherigen Unterabschnitt beschriebenen Fehler und Wiederholungen auch bei der Kompensation vorhanden sind. Andererseits muss die Kompensation auch eine "Nullkompensation" berücksichtigen, da die Vorwärtsoperation von Saga einen Fehler zurückgeben kann, der vor oder nach der Datenanpassung aufgetreten sein kann. Bei Fehlern, bei denen die Anpassung vorgenommen wurde, müssen wir die umgekehrte Anpassung durchführen. aber für Fehler, bei denen die Anpassung nicht festgeschrieben wurde, müssen wir die umgekehrte Operation überspringen.

In der von DTM bereitgestellten Hilfstabelle und Hilfsfunktionen wird einerseits bestimmt, ob die Kompensation eine Nullkompensation ist, basierend auf der durch die Vorwärtsoperation eingefügten Gid, und andererseits wird Gid+'compensate' erneut eingefügt um festzustellen, ob es sich bei der Kompensation um einen doppelten Vorgang handelt. Wenn es einen normalen Kompensationsvorgang gibt, führt es die Datenanpassung für das Geschäft durch; Wenn es keine oder eine doppelte Vergütung gibt, wird die Anpassung für das Unternehmen übersprungen.

Der MySQL-Code lautet wie folgt.

app.POST(BusiAPI+"/SagaBTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error {
        return SagaAdjustBalance(tx, TransInUID, -reqFrom(c).Amount, "")
    })
}))

Der Code für Redis lautet wie folgt.

app.POST(BusiAPI+"/SagaRedisTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), reqFrom(c).Amount, 7*86400)
}))

Der Kompensationsdienstcode ist fast identisch mit dem vorherigen Code der Vorwärtsoperation, außer dass der Betrag mit -1 multipliziert wird. Die DTM-Hilfsfunktion handhabt Idempotenz und Nullkompensation automatisch richtig.

Andere Ausnahmen

Beim Schreiben von Termingeschäften und Kompensationsgeschäften gibt es tatsächlich eine weitere Ausnahme namens „Suspension“. Eine globale Transaktion wird zurückgesetzt, wenn das Zeitlimit abgelaufen ist oder die Wiederholungsversuche das konfigurierte Limit erreicht haben. Der Normalfall ist, dass die Vorwärtsoperation vor der Kompensation durchgeführt wird, aber im Falle einer Prozessunterbrechung kann die Kompensation vor der Vorwärtsoperation durchgeführt werden. Daher muss die Vorwärtsoperation auch feststellen, ob die Kompensation ausgeführt wurde, und falls dies der Fall ist, muss die Datenanpassung ebenfalls übersprungen werden.

Für DTM-Benutzer wurden diese Ausnahmen elegant und ordnungsgemäß gehandhabt, und Sie als Benutzer müssen nur dem MustBarrierFromGin(c).Call folgen wie oben beschrieben anrufen und brauchen sich überhaupt nicht darum zu kümmern. Das Prinzip der DTM-Behandlung dieser Ausnahmen wird hier ausführlich beschrieben:Ausnahmen und Teiltransaktionsbarrieren

Initiieren einer verteilten Transaktion

Nach dem Schreiben der einzelnen Teiltransaktionsdienste initiiert der folgende Code des Codes eine globale Saga-Transaktion.

saga := dtmcli.NewSaga(dtmutil.DefaultHTTPServer, dtmcli.MustGenGid(dtmutil.DefaultHTTPServer)).
  Add(busi.Busi+"/SagaBTransOut", busi.Busi+"/SagaBTransOutCom", &busi.TransReq{Amount: 50}).
  Add(busi.Busi+"/SagaMongoTransIn", busi.Busi+"/SagaMongoTransInCom", &busi.TransReq{Amount: 30}).
  Add(busi.Busi+"/SagaRedisTransIn", busi.Busi+"/SagaRedisTransOutIn", &busi.TransReq{Amount: 20})
err := saga.Submit()

In diesem Teil des Codes wird eine globale Saga-Transaktion erstellt, die aus 3 Untertransaktionen besteht.

• 50 von Mysql übertragen
• Umsteigen in 30 nach Mongo
• Transfer in 20 zu Redis

Wenn während der gesamten Transaktion alle Teiltransaktionen erfolgreich abgeschlossen werden, ist die globale Transaktion erfolgreich; Wenn eine der Untertransaktionen einen Geschäftsausfall zurückgibt, wird die globale Transaktion zurückgesetzt.

Lauf

Wenn Sie ein vollständiges Beispiel des Obigen ausführen möchten, gehen Sie wie folgt vor.

1. DTM ausführen

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

1. Führen Sie ein erfolgreiches Beispiel durch

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_multidb

1. Führen Sie ein fehlgeschlagenes Beispiel aus

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_multidb_rollback

Sie können das Beispiel ändern, um verschiedene vorübergehende Fehler, Nullkompensationssituationen und verschiedene andere Ausnahmen zu simulieren, bei denen die Daten konsistent sind, wenn die gesamte globale Transaktion abgeschlossen ist.

Zusammenfassung

Dieser Artikel gibt ein Beispiel für eine verteilte Transaktion über Mysql, Redis und Mongo. Es beschreibt detailliert die Probleme, die behandelt werden müssen, und die Lösungen.

Die Prinzipien in diesem Artikel sind für alle Speicher-Engines geeignet, die ACID-Transaktionen unterstützen, und Sie können sie schnell für andere Engines wie TiKV erweitern.

Willkommen auf github.com/dtm-labs/dtm. Es ist ein spezielles Projekt, um verteilte Transaktionen in Microservices einfacher zu machen. Es unterstützt mehrere Sprachen und mehrere Muster wie eine 2-Phasen-Nachricht, Saga, Tcc und Xa.