マイクロサービスにSagaパターンを実装する方法

マイクロサービスアーキテクチャは、アプリケーションサービスとそのデータベースを別々に維持することができます.シーケンスとして、それは複数のサービスを横切って伝播する必要があるデータ変化を持っているのが一般的です.データベーストランザクションは、複数のサービスを横断することはできませんので、データの整合性の課題になります.

佐賀

佐賀は、この論文に記載されている分散トランザクションパターンですSAGAS . コア・アイデアは複数の短いローカルトランザクションに分割することです.そして、それはサガトランザクションコーディネーターによって調整されます、その結果、各々のローカルトランザクションが首尾よく完了するならば、それは正常に完了します、そして、誰かが失敗するならば、補償操作は逆の順序で一度に1つを起動されます.
この記事は、読者の佐賀取引の正確な理解を与えるために完全な佐賀の例を提示します

ビジネスシナリオ

銀行間移動は典型的な分配されたトランザクションシナリオです.そこで、銀行に銀行を通してお金を移す必要があります.

佐賀取引

私たちはあなたに詳細な実行可能な例ベースを提示しますdtm , 分散トランザクション・フレームワーク
あなたが転送とロールバックのためにビジネスのあなたのインプリメンテーションを終えたと仮定してください.

"/api/sagabtransout "

トランスフォーメーションのための"API/SagabTransoutcom "補償

"/API/サグアベンキン"

トランスに対する"API/Sagabtransincom "補償

次のコードは、これらの操作をSagas分散トランザクションに調整します.両方のトランザクションが終了するTransIn and TransOut 成功するか、両方ともロールバックします.いずれの場合も、AとBのバランスの和は同じままである.

    req := &gin.H{"amount": 30} // load of microservice
    // DtmServer is the address of the DTM service
    saga := dtmcli.NewSaga(DtmServer, dtmcli.MustGenGid(DtmServer)).
        // Add a child transaction of TransOut with url: qsBusi+"/TransOut" for the forward operation and url: qsBusi+"/TransOutCom" for the reverse operation
        Add(qsBusi+"/SagaBTransOut", qsBusi+"/SagaBTransOutCom", req).
        // Add a subtransaction of TransIn with url: qsBusi+"/TransOut" for the forward action and url: qsBusi+"/TransInCom" for the reverse action
        Add(qsBusi+"/SagaBTransIn", qsBusi+"/SagaBTransInCom", req)
    // commit saga transaction, dtm will complete all subtransactions or rollback all subtransactions
    err := saga.Submit()

成功したトランザクションタイミング図は次の通りです.

コアオペレーション

ユーザーバランスの調整と補償は慎重に扱われるべきです.ここでは、調整の詳細に飛び込む.銀行振替の例では、我々は実行するつもりですTransOut and TransIn なお、補正動作においては、作用動作及び逆調整においても同様である.
まず、アカウントのバランス表を作成します.

CREATE TABLE dtm_busi.`user_account` (
  `id` int(11) AUTO_INCREMENT PRIMARY KEY,
  `user_id` int(11) not NULL UNIQUE ,
  `balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `trading_balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `create_time` datetime DEFAULT now(),
  `update_time` datetime DEFAULT now()
);

次に、ユーザーのアカウント残高を調整するコアビジネスコードを書く

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

次に、アクション/補償操作のための特定の処理関数を書きます

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

これらの処理機能のコアロジックは、バランスを調整することですbarrier.Call 詳細は後述する

ラン

これらの手順に従って、成功した例を実行します.

分散トランザクションを管理するDTMを実行する

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

例をあげる

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

ネットワーク例外の処理

DTMにコミットされたトランザクションが、オペレーションが起動されたときに一時的な障害を持つとします.DTMは不完全な操作を再試行します.そして、それはグローバルなトランザクションの下位トランザクションを必要とします.DTMフレームワークは、サブトランザクションバリア技術を開拓しました.機能を提供するCall これにより、この関数の内部での操作は、一度に最も頻繁に行われるようになります.

func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error

このブランチバリアは、自動的に無効化だけでなく、null補正とハングアップ問題を扱うことができますexceptions and solutions 詳細は

ロールバックハンドリング

銀行がユーザーBに量を移す準備をしていて、ユーザーBのアカウントが異常であるとわかって、失敗を返すならば、何が起こりますか?転送操作が失敗を返すように、ハンドラ関数を更新します

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  return dtmcli.ErrFailure
}))

トランザクション故障相互作用のタイミング図を与えた

Translin支店の行動は何もしなかった.トランスイン枝の補償は、逆の調整が間違っている原因になりますか?
心配しないで、前のサブトランザクションバリア技術は、エラーがコミットの前に起こるなら、トランスイン障害がヌル操作として補償されるのを確実にして、コミットの後、エラーが起こるならば、逆の調整をするために補償されます
コミット後にエラーを返すトランスフォームを変更できます.

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

最終的な結果の残高はまだ正しいでしょうExceptions and Solutions 詳細は

概要

この記事では、完全なSAGAトランザクションソリューションを、この例にいくつかの簡単な変更を使用して実際の問題を解決するために使用することができます作業佐賀
佐賀の詳細はこちらSAGA
あなたは訪問を歓迎されますhttps://github.com/dtm-labs/dtm