PythonでSaga分散トランザクションを簡単に完了する

[[422283]]

銀行間送金業務は、典型的な分散型トランザクションのシナリオです。 A が銀行間で B に送金する必要がある場合、2 つの銀行のデータが関係します。転送の ACID は、データベースのローカルトランザクションでは保証できず、分散トランザクションを通じてのみ解決できます。

分散トランザクション

分散環境では、分散トランザクションは BASE 理論に従って、可用性、パフォーマンス、および劣化したサービスのニーズを満たし、一貫性と分離の要件を軽減します。

基本的な可用性
ソフトステート
最終的な一貫性

一方、分散トランザクションも ACID 仕様に部分的に準拠しています。

原子性: 厳密に従う
一貫性: 取引が完了した後の一貫性は厳密に遵守されます。取引中の一貫性は適切に緩和できる
分離: 並列トランザクションは相互に影響を与えることができません。中間取引結果の可視性により、安全にリラックスできる
粘り強さ: 厳守

サガ

Saga は、データベース論文 SAGAS に記載されている分散トランザクションソリューションです。基本的な考え方は、長いトランザクションを複数のローカルの短いトランザクションに分割し、それらを Saga トランザクションコーディネータが調整することです。各ローカルトランザクションが正常に完了した場合、正常に完了します。ステップが失敗した場合、補正操作が逆の順序で 1 回呼び出されます。

現在、SAGA に使用できるオープンソースフレームワークは主に Java 言語であり、その中でも seata が代表的です。この例では Go 言語を使用し、使用される分散トランザクションフレームワークは https://github.com/yedf/dtm です。これは分散トランザクションを非常にエレガントにサポートします。 SAGA の構成を詳しく説明しましょう。

DTM トランザクションフレームワークには、従来の XA 分散トランザクションと同じように、次の 3 つのロールがあります。

AP/アプリケーションは、グローバルトランザクションを開始し、どのトランザクションブランチがグローバルトランザクションに含まれるかを定義します。
支店業務のさまざまなリソースの管理を担当するRM/リソースマネージャー
TM/トランザクションマネージャは、SAGA フォワード/リバース操作の実行を含むグローバルトランザクションの正しい実行を調整する役割を担います。

正常に完了した SAGA シーケンス図を見てみましょう。SAGA 分散トランザクションを簡単に理解できます。

SAGAの実践

銀行振込の例では、フォワード操作で振込と振出を行い、補正操作で反対の調整を行います。

まず、口座残高テーブルを作成します。

テーブル dtm_busi.`user_account` を作成します (
  `id` int ( 11 ) AUTO_INCREMENT 主キー、
  `user_id` int ( 11 ) NULLではない一意、
  `balance` 小数点( 10 , 2 ) NOT NULL デフォルト'0.00' ,
  `create_time` datetime DEFAULT now(),
  `update_time` 日時 DEFAULT now()
 );

まずはユーザーの口座残高を調整するためのコアビジネスコードを書いてみましょう

def saga_adjust_balance(カーソル、uid、量):
  影響を受ける = utils.sqlexec(cursor, "dtm_busi.user_account を更新し、残高をbalance+%d に設定し、user_id=%d かつ balance >= -%d" %(amount, uid, amount))
影響を受ける場合== 0 :
    例外が発生します( 「更新エラー、残高が足りません」 )

次に、具体的な順方向演算・補正演算処理関数を書いてみます。

 @app .post( "/api/TransOutSaga" )
 trans_out_saga() を定義します:
  saga_adjust_balance(c, out_uid, - 30 )
戻り値: "dtm_result" : "SUCCESS" }
 @app .post( "/api/TransOutCompensate" )
 trans_out_compensate()を定義します:
  saga_adjust_balance(c, out_uid, 30 ) は、
戻り値: "dtm_result" : "SUCCESS" }
 @app .post( "/api/TransInSaga" )
 trans_in_saga() を定義します:
  saga_adjust_balance(c, in_uid, 30 ) は、
戻り値: "dtm_result" : "SUCCESS" }
 @app .post( "/api/TransInCompensate" )
トランスイン補償()を定義します:
  saga_adjust_balance(c, in_uid, - 30 )
戻り値: "dtm_result" : "SUCCESS" }

この時点で、各サブトランザクションの処理関数は正常であり、次にSAGAトランザクションが開かれ、分岐呼び出しが行われます。

 # これはDTMサービスのアドレスです
dtm = "http://localhost:8080/api/dtmsvr"
 # これはビジネスマイクロサービスのアドレスです
svc = "http://localhost:5000/api"
    必要数 = { "金額" : 30 }
    s = saga.Saga(dtm、utils.gen_gid(dtm))
    s.add(req, svc + "/TransOutSaga" 、 svc + "/TransOutCompensate" )
    s.add(req, svc + "/TransInSaga" 、 svc + "/TransInCompensate" )
    s.送信()

この時点で、完全な SAGA 分散トランザクションが記述されました。

完全な成功例を実行したい場合は、この例yedf/dtmcli-py-sampleを参照してください。実行は非常に簡単です。

 # dtmを展開して起動する
# docker バージョン18以上が必要です
gitクローンhttps://github.com/yedf/dtm
 CDDTM
 docker-compose を起動する
# 別のコマンドラインを開始する
gitクローンhttps://github.com/yedf/dtmcli-py-sample
 cd dtmcli-py-サンプル
pip3 インストール flask dtmcli リクエスト
フラスコラン
# 別のコマンドラインを開始する
カールローカルホスト: 5000 /api/fireSaga

ネットワーク異常の処理

dtm に送信されたトランザクションで転送操作を呼び出すときに短時間の障害が発生した場合はどうなりますか? SAGA トランザクションプロトコルに従って、dtm は未完了の操作を再試行します。こういう時、私たちは何をすべきでしょうか？障害は、転送操作の完了後にネットワーク障害が発生したか、転送操作中にマシンがクラッシュした可能性があります。口座残高の調整が正しいことをどのように確認すればよいでしょうか?

このようなネットワーク異常を適切に処理することは、分散トランザクションにおける大きな問題です。異常には、重複リクエスト、空補償、停止の 3 種類があり、すべて適切に処理する必要があります。

DTM は、上記の異常な状況におけるビジネスロジックが正しい順序で 1 回だけ正常に送信されるようにするためのサブトランザクションバリア機能を提供します。 (サブトランザクションバリアの詳細については、分散トランザクションの最も古典的な 7 つのソリューションのサブトランザクションバリアのセクションを参照してください)

処理機能を次のように調整します。

 @app .post( "/api/TransOutSaga" )
 trans_out_saga() を定義します:
  カーソルとしてbarrier.AutoCursor(conn_new())を使用します:
    busi_callback(c)を定義します。
      saga_adjust_balance(c, out_uid, - 30 )
    バリアー_from_req(リクエスト).call(カーソル, busi_callback)
戻り値: "dtm_result" : "SUCCESS" }

ここでのbarrier_from_req(request).call(cursor, busi_callback)呼び出しは、サブトランザクションバリア技術を使用して、busi_callbackコールバック関数が1回だけ送信されるようにします。

TransIn サービスを複数回呼び出すことができますが、残高は 1 回だけ調整されます。

ロールバックの処理

銀行がユーザー 2 の口座に異常があることを発見し、ユーザー 2 に金額を振り込もうとしたときに失敗を返した場合、どうなるでしょうか。転送操作が失敗を返すように処理機能を調整します

@app .post( "/api/TransInSaga" )
 trans_in_saga() を定義します:
戻り値: "dtm_result" : "FAILURE" }

トランザクション失敗の相互作用のタイミング図を示します。

ここに一つあります。 TransIn の転送操作は何も実行されず、失敗を返しました。このとき、TransIn の補正操作を呼び出すと、逆調整がうまくいかなくなるのでしょうか?

心配しないでください。以前のサブトランザクションバリアテクノロジにより、送信前に TransIn エラーが発生した場合でも、補償は何も行われないことが保証されます。送信後に TransIn エラーが発生した場合、補正操作によってデータが 1 回送信されます。

エラーを返す TransIn を次のように変更できます。

 @app .post( "/api/TransInSaga" )
 trans_in_saga() を定義します:
  カーソルとしてbarrier.AutoCursor(conn_new())を使用します:
    busi_callback(c)を定義します。
      saga_adjust_balance(c, in_uid, 30 ) は、
    バリアー_from_req(リクエスト).call(カーソル, busi_callback)
戻り値: "dtm_result" : "FAILURE" }

最終結果のバランスは依然として正しいものになります。この原則は、分散トランザクションの最も古典的な7つのソリューションのサブトランザクションバリアリンクに記載されています。

まとめ

この記事では、SAGA の理論的知識を紹介し、通常の正常完了、異常な状況、および正常なロールバックを含む SAGA トランザクションの作成プロセスの完全な例を示しました。この記事を通じて読者の皆様にはSAGAへの深い理解を頂けたのではないかと思います。

この記事で使用されている dtm は、強力な機能を備えた新しいオープンソースの Golang 分散トランザクション管理フレームワークです。 TCC、SAGA、XAなどのトランザクションモード、トランザクションメッセージをサポートし、Go、Python、PHP、Node、CSharpなどの言語をサポートします。同時に、非常にシンプルで使いやすいインターフェースも提供します。

この記事を読んだ後は、ぜひプロジェクト https://github.com/yedf/dtm にアクセスし、スターを付けてサポートしてください。

<<: Kubernetesがサービスディスカバリを実装する方法について話す

>>: CentOS は、kafka にログを収集するために filebeat をインストールします