Как да внедрите разпределена транзакция в Mysql, Redis и Mongo

Mysql, Redis и Mongo са много популярни магазини и всеки има своите предимства. В практическите приложения е обичайно да се използват няколко магазина едновременно и осигуряването на съгласуваност на данните в множество магазини става изискване.

Тази статия дава пример за внедряване на разпределена транзакция в множество двигатели на магазина, Mysql, Redis и Mongo. Този пример се основава на рамката за разпределени транзакции https://github.com/dtm-labs/dtm и се надяваме, че ще помогне за решаването на проблемите ви с последователността на данните в микроуслугите.

Възможността за гъвкаво комбиниране на множество механизми за съхранение за формиране на разпределена транзакция е първо предложена от DTM и никоя друга рамка за разпределени транзакции не е заявила такава възможност.

Проблемни сценарии

Нека първо разгледаме проблемния сценарий. Да предположим, че потребител сега участва в промоция:той или тя има баланс, презарежда телефонната сметка и промоцията ще раздаде точки в мола. Балансът се съхранява в Mysql, сметката се съхранява в Redis, точките на мола се съхраняват в Mongo. Тъй като промоцията е ограничена във времето, има вероятност участието да не успее, така че е необходима поддръжка за връщане.

За горния проблемен сценарий можете да използвате транзакцията на DTM Saga и ние ще обясним подробно решението по-долу.

Подготовка на данните

Първата стъпка е да подготвите данните. За да улесним потребителите бързо да започнат работа с примерите, ние сме подготвили съответните данни на en.dtm.pub, които включват Mysql, Redis и Mongo, а конкретното потребителско име и парола за връзка могат да бъдат намерени на https:// github.com/dtm-labs/dtm-examples.

Ако искате сами да подготвите средата за данни локално, можете да използвате https://github.com/dtm-labs/dtm/blob/main/helper/compose.store.yml, за да стартирате Mysql, Redis, Mongo; и след това изпълнете скриптове в https://github.com/dtm-labs/dtm/tree/main/sqls, за да подготвите данните за този пример, където busi.* е бизнес данните и barrier.* е спомагателната маса, използвана от DTM

Написване на бизнес код

Нека започнем с бизнес кода за най-познатия Mysql.

Следният код е на Голанг. Други езици като C#, PHP, Java можете да намерите тук:DTM SDK

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
}

Този код основно извършва корекция на баланса на потребителя в базата данни. В нашия пример тази част от кода се използва не само за операция напред на Saga, но и за операцията за компенсация, където трябва да се предаде само отрицателна сума за компенсация.

За Redis и Mongo бизнес кодът се обработва по подобен начин, просто се увеличават или намаляват съответните салда.

Как да осигурим идемпотентност

За модела на транзакциите на Saga, когато имаме временен неуспех в услугата за подтранзакция, неуспешната операция ще бъде изпробвана отново. Тази грешка може да възникне преди или след извършване на подтранзакцията, така че операцията по подтранзакцията трябва да бъде идемпотентна.

DTM предоставя помощни таблици и помощни функции, за да помогне на потребителите бързо да постигнат идемпотентност. За Mysql ще създаде помощна таблица barrier в бизнес базата данни, когато потребителят започне транзакция за коригиране на баланса, той първо ще вмъкне Gid в barrier маса. Ако има дублиран ред, тогава вмъкването ще бъде неуспешно и след това пропуснете корекцията на баланса, за да осигурите идемпотент. Кодът, използващ помощната функция, е както следва:

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 се справя с идемпотентността по подобен начин на Mysql, така че няма да навлизам отново в подробности.

Redis се справя с идемпотентността по различен начин от Mysql, главно поради разликата в принципа на транзакциите. Транзакциите на Redis се осигуряват главно чрез атомно изпълнение на Lua. помощната функция на DTM ще регулира баланса чрез Lua скрипт. Преди да коригира баланса, той ще поиска Gid в Редис. Ако Gid съществува, ще пропусне корекцията на баланса; ако не, ще запише Gid и извършете корекцията на баланса. Кодът, използван за помощната функция, е както следва:

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

Как да направя компенсация

За Saga също трябва да се справим с операцията за компенсация, но компенсацията не е просто обратна корекция и има много подводни камъни, които трябва да знаете.

От една страна, компенсацията трябва да вземе предвид идемпотентността, тъй като неуспехът и повторните опити, описани в предишния подраздел, също съществуват при компенсацията. От друга страна, компенсацията също трябва да вземе предвид "нулевата компенсация", тъй като операцията напред на Saga може да върне грешка, която може да се е случила преди или след корекцията на данните. За неизправности, при които е извършена корекцията, трябва да извършим обратната настройка; но при неизправности, при които корекцията не е извършена, трябва да пропуснем обратната операция.

В помощната таблица и помощните функции, предоставени от DTM, от една страна, той ще определи дали компенсацията е нулева компенсация въз основа на Gid, въведена от операцията напред, а от друга страна, ще вмъкне Gid+ „компенсиране“ отново за да се определи дали компенсацията е дублирана операция. Ако има нормална компенсационна операция, тогава тя ще изпълни корекцията на данните за бизнеса; ако има нулева компенсация или дублирана компенсация, тя ще пропусне корекцията на бизнеса.

Кодът на Mysql е както следва.

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, "")
    })
}))

Кодът за Redis е както следва.

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

Кодът на услугата за компенсация е почти идентичен с предишния код на операцията напред, с изключение на това, че сумата се умножава по -1. Помощната функция на DTM автоматично обработва правилно идемпотентността и нулевата компенсация.

Други изключения

При писане на пренасочващи операции и операции за компенсация всъщност има друго изключение, наречено „Преустановяване“. Глобална транзакция ще се върне назад, когато изтече времето за изчакване или повторните опити достигнат конфигурирания лимит. Нормалният случай е, че операцията напред се извършва преди компенсацията, но в случай на спиране на процеса компенсацията може да се извърши преди операцията напред. Така че операцията напред също трябва да определи дали компенсацията е била изпълнена, а в случай, че е, корекцията на данните също трябва да бъде пропусната.

За потребителите на DTM тези изключения са обработени елегантно и правилно и вие, като потребител, трябва само да следвате MustBarrierFromGin(c).Call обадете се, описано по-горе и изобщо не е нужно да се интересувате от тях. Принципът за DTM обработка на тези изключения е описан подробно тук:Изключения и бариери за подтранзакции

Иницииране на разпределена транзакция

След написването на отделните услуги за подтранзакция, следните кодове на кода инициират глобална транзакция на Saga.

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()

В тази част от кода се създава глобална транзакция на Saga, която се състои от 3 подтранзакции.

• Прехвърлете 50 от Mysql
• Прехвърляне след 30 до Монго
• Прехвърляне след 20 до Redis

По време на транзакцията, ако всички подтранзакции завършат успешно, тогава глобалната транзакция е успешна; ако една от подтранзакциите върне бизнес неуспех, тогава глобалната транзакция се връща назад.

Изпълнете

Ако искате да изпълните пълен пример от горното, стъпките са както следва.

1. Изпълнете DTM

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

1. Изпълнете успешен пример

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

1. Изпълнете неуспешен пример

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

Можете да модифицирате примера, за да симулирате различни временни неуспехи, ситуации с нулева компенсация и различни други изключения, при които данните са последователни, когато цялата глобална транзакция приключи.

Резюме

Тази статия дава пример за разпределена транзакция в Mysql, Redis и Mongo. В него се описват подробно проблемите, които трябва да бъдат решени, и решенията.

Принципите в тази статия са подходящи за всички машини за съхранение, които поддържат ACID транзакции и можете бързо да го разширите за други машини като TiKV.

Добре дошли да посетите github.com/dtm-labs/dtm. Това е специален проект за улесняване на разпределените транзакции в микроуслугите. Той поддържа множество езици и множество модели, като двуфазно съобщение, Saga, Tcc и Xa.