GOでSQLデータベースを操作する場合、マルチステップトランザクション中の原子性とロールバックの管理が困難になる可能性があります。この記事では、GOでSQLトランザクションを実行するための堅牢で再利用可能でテスト可能なフレームワークを作成し、柔軟性のためにジェネリックを使用して、

を使用してガイドします。

トランザクション内で複数の従属データベース操作を実行するためのSQLWriteExecユーティリティを構築します。ステートレスとステートの両方の運用の両方をサポートし、依存関係をシームレスに管理しながら、関連するエンティティを挿入するなどの洗練されたワークフローを可能にします。

SQLトランザクションのフレームワークが必要なのはなぜですか？

実際のアプリケーションでは、データベース操作が隔離されることはめったにありません。これらのシナリオを考慮してください：

ユーザーを挿入し、インベントリを原子的に更新します。
注文を作成して支払いを処理し、一貫性を確保します。
複数のステップが関与すると、障害中にロールバックを管理することが重要になり、データの整合性を確保するために重要になります。

TXNマネジメントでGOを使用します。

データベースTXNを書いている場合、コアロジックを書く前に考慮する必要があるボイラープレートがいくつかある可能性があります。このTXN管理はJavaのSpring Bootによって管理されていますが、Javaでコードを書いている間、あなたはそれらについてあまり気にすることはありませんでしたが、Golangではそうではありません。以下に簡単な例を以下に示します。

func basictxn（db *sql.db）エラー{ //トランザクションを開始します Tx、err：= db.begin（） err！= nil { errを返します } defer func（）{ r：= recover（）; r！= nil { tx.rollback（） } elseの場合！= nil { tx.rollback（） } それ以外 { tx.commit（） } }（） //注文テーブルにデータを挿入します _、err = tx.exec（ "注文の挿入（id、customer_name、order_date）values（1、 'john doe'、 '2022-01-01'）"）） err！= nil { errを返します } nilを返します }

func basicTxn(db *sql.DB) error {
    // start a transaction
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        } else if err != nil {
            tx.Rollback()
        } else {
            tx.Commit()
        }
    }()

    // insert data into the orders table
    _, err = tx.Exec("INSERT INTO orders (id, customer_name, order_date) VALUES (1, 'John Doe', '2022-01-01')")
    if err != nil {
        return err
    }
    return nil
}

私は後の選択をしましたが、コードの変更は以下に見ることができます。

func testsqlwriteexec_createordertxn（t *testing.t）{ db：= setupdatabase（） //新しいSQL Write Executorを作成します err：= dbutils.newsqltxnexec [OrderRequest、OrderProcessingResponse]（Context.todo（）、db、nil、&orderRequest {customername： "customera"、productid：1、数量：10}）。 StateFulfexec（InsertOrder）。 StateFulexec（UpdateInventory）。 StateFulfexec（挿入船）。 専念（） //トランザクションが正常にコミットされたかどうかを確認します err！= nil { T.ファタル（err） 戻る } verifytransactionsuccessful（t、db） T.Cleanup（ func（）{ クリーンアップ（DB） db.close（） }、 ）） }

func insertordord（ctx context.context、txn *sql.tx、order *orderrequest、orderprocessing *orderprocessingresponse）error { //注文を挿入します 結果、err：= txn.exec（ "注文の挿入（customer_name、product_id、数量）値（$ 1、$ 2、$ 3）"、order.customername、order.productid、order.quantity） err！= nil { errを返します } //挿入された注文IDを取得します OrderProcessing.Orderid、err = result.lastinsertid（） errを返します } func updateInventory（ctx context.context、txn *sql.tx、order *orderRequest、orderprocessing *orderprocessingresponse）error { //在庫が存在し、数量が存在する場合は数量チェックよりも大きい場合は在庫を更新します 結果、err：= txn.exec（ "inventory set product_quantity = product_quantity- $ 1ここで、id = $ 2およびproduct_quantity> = $ 1"、order.quantity、order.productid） err！= nil { errを返します } //影響を受ける行の数を取得します nowsaffected、err：= result.rowsachpected（） rowsaffected == 0の場合{ return errors.new（ "在庫不足"） } errを返します } func insertshipment（ctx context.context、txn *sql.tx、order *orderrequest、orderprocessing *orderprocessingresponse）error { //出荷を挿入します 結果、err：= txn.exec（ "shipping_info（customer_name、shipping_address）values（$ 1、 '配送先住所'）"、order.customername）values（customer_name、shipping_address） err！= nil { errを返します } //挿入された配送IDを取得します OrderProcessing.ShippingId、err = result.lastinsertid（） errを返します }

func TestSqlWriteExec_CreateOrderTxn(t *testing.T) {

    db := setupDatabase()
    // create a new SQL Write Executor
    err := dbutils.NewSqlTxnExec[OrderRequest, OrderProcessingResponse](context.TODO(), db, nil, &OrderRequest{CustomerName: "CustomerA", ProductID: 1, Quantity: 10}).
        StatefulExec(InsertOrder).
        StatefulExec(UpdateInventory).
        StatefulExec(InsertShipment).
        Commit()
    // check if the transaction was committed successfully
    if err != nil {
        t.Fatal(err)
        return
    }
    verifyTransactionSuccessful(t, db)
    t.Cleanup(
        func() { 
            cleanup(db)
            db.Close() 
        },
    )
}

このコードは非常に正確で簡潔になります。

func InsertOrder(ctx context.Context, txn *sql.Tx, order *OrderRequest, orderProcessing *OrderProcessingResponse) error {
    // Insert Order
    result, err := txn.Exec("INSERT INTO orders (customer_name, product_id, quantity) VALUES ($1, $2, $3)", order.CustomerName, order.ProductID, order.Quantity)
    if err != nil {
        return err
    }
    // Get the inserted Order ID
    orderProcessing.OrderID, err = result.LastInsertId()
    return err
}

func UpdateInventory(ctx context.Context, txn *sql.Tx, order *OrderRequest, orderProcessing *OrderProcessingResponse) error {
    // Update Inventory if it exists and the quantity is greater than the quantity check if it exists
    result, err := txn.Exec("UPDATE inventory SET product_quantity = product_quantity - $1 WHERE id = $2 AND product_quantity >= $1", order.Quantity, order.ProductID)
    if err != nil {
        return err
    }
    // Get the number of rows affected
    rowsAffected, err := result.RowsAffected()
    if rowsAffected == 0 {
        return errors.New("Insufficient inventory")
    }
    return err
}

func InsertShipment(ctx context.Context, txn *sql.Tx, order *OrderRequest, orderProcessing *OrderProcessingResponse) error {
    // Insert Shipment
    result, err := txn.Exec("INSERT INTO shipping_info (customer_name, shipping_address) VALUES ($1, 'Shipping Address')", order.CustomerName)
    if err != nil {
        return err
    }
    // Get the inserted Shipping ID
    orderProcessing.ShippingID, err = result.LastInsertId()
    return err
}

コアロジックの実装方法

アイデアは、複数のTXNを受け入れることができるように、TXNを単一のGO構造体に分離することです。 TXNとは、クラス用に作成したTXNでアクションを実行する機能を意味します。

タイプTxnfn [t Any] func（ctx context.context、txn *sql.tx、processreq *t）エラー タイプStateFultXnfn [t any、r any] func（ctx context.context、txn *sql.tx、processreq *t、processedres *r）エラー

これら2つは、何かを処理するためにTXNを取り入れる関数タイプです。次に、このような関数を作成し、ARGSを注入して関数を実行する執行者クラスに渡すデータレイヤーで。

type TxnFn[T any] func(ctx context.Context, txn *sql.Tx, processingReq *T) error
type StatefulTxnFn[T any, R any] func(ctx context.Context, txn *sql.Tx, processingReq *T, processedRes *R) error

// sql書き込み操作を実行するときに責任を負う //依存書の書き込みについては、依存データをProcessReqに追加し、次の関数呼び出しに進む必要がある場合があります タイプsqltxnexec [t any、r any] struct { db *sql.db txn *sql.tx txnfns [] txnfn [t] statefultxnfns [] statefultxnfn [t、r] processreq *t ProcessedRes *r ctx context.context エラーエラー }

これは、すべてのTXN_FNの詳細を保存する場所であり、TXNをコミットするためのcommit（）方法があります。

// SQL Write Executor is responsible when executing write operations
// For dependent writes you may need to add the dependent data to processReq and proceed to the next function call
type SqlTxnExec[T any, R any] struct {
    db               *sql.DB
    txn              *sql.Tx
    txnFns         []TxnFn[T]
    statefulTxnFns []StatefulTxnFn[T, R]
    processingReq    *T
    processedRes     *R
    ctx              context.Context
    err              error
}

func（s *sqltxnexec [t、r]）commit（）（err erry）{ defer func（）{ p：= recover（）; p！= nil { s.txn.rollback（） パニック（P） } elseの場合！= nil { err = errors.join（err、s.txn.rollback（）） } それ以外 { err = errors.join（err、s.txn.commit（）） } 戻る }（） _、writefn：= range s.txnfns { err = writefn（s.ctx、s.txn、s.processingreq）; err！= nil { 戻る } } _、statefulwritefn：= range s.statefultxnfns { err = statefulwritefn（s.ctx、s.txn、s.processingreq、s.processedres）; err！= nil { 戻る } } 戻る }

レポでより多くの例とテストを見つけることができます -

func (s *SqlTxnExec[T, R]) Commit() (err error) {
    defer func() {
        if p := recover(); p != nil {
            s.txn.Rollback()
            panic(p)
        } else if err != nil {
            err = errors.Join(err, s.txn.Rollback())
        } else {
            err = errors.Join(err, s.txn.Commit())
        }
        return
    }()

    for _, writeFn := range s.txnFns {
        if err = writeFn(s.ctx, s.txn, s.processingReq); err != nil {
            return
        }
    }

    for _, statefulWriteFn := range s.statefulTxnFns {
        if err = statefulWriteFn(s.ctx, s.txn, s.processingReq, s.processedRes); err != nil {
            return
        }
    }
    return
}

最近では、分散システムとコンセンサスプロトコルにバイアスをかけていますが、それでもSQLを使用していますが、
は存在します。

誰かがこの上に貢献して構築したいかどうかを教えてください!!

ここまで読んでくれてありがとう!!

https://in.linkedin.com/in/mahadev-k-93452023

https://x.com/mahadev_k_