マイクロサービスのトランザクション: コレオグラフィーを使用したパート SAGA パターン-Golang-php.cn

このシリーズの最初の記事では、SAGA パターンを紹介し、最小限の オーケストレーション が中央オーケストレーターを使用して分散トランザクションを管理する方法を示しました。

実際にやってみましょう!今回は、サービスが自律的にイベントを発行および消費することでワークフローを調整する コレオグラフィーアプローチについて詳しく説明します。

これを実用化するために、Go と RabbitMQ を使用してマルチサービスのヘルスケアワークフローを実装します。各サービスには独自の main.go があり、スケール、テスト、独立した実行が容易になります。

SAGAコレオグラフィーとは何ですか？

振り付けは分散型コミュニケーションに依存しています。各サービスはイベントをリッスンし、新しいイベントを発行することで後続のステップをトリガーします。中央のオーケストレーターは存在しません。フローは個々のサービスの相互作用から生まれます。

主な利点:

分離されたサービス: 各サービスは独立して動作します。
スケーラビリティ: イベント駆動型システムは高負荷を効率的に処理します。
柔軟性: 新しいサービスを追加する場合、ワークフローロジックを変更する必要はありません。

課題:

デバッグの複雑さ: 複数のサービスにわたるイベントの追跡は難しい場合があります。 (このトピックに特化した記事を書きます。お楽しみに!)
インフラストラクチャのセットアップ: サービスには、すべての点を接続するための堅牢なメッセージブローカー (RabbitMQ など) が必要です。
イベントストーム: ワークフローの設計が不十分だと、システムがイベントで圧倒される可能性があります。

実践例: 医療ワークフロー

最初の記事のヘルスケアワークフローをもう一度見てみましょう:

患者サービス: 患者の詳細と保険適用範囲を確認します。
スケジューラサービス: 手順をスケジュールします。
在庫サービス: 医療用品を予約します。
請求サービス: 請求を処理します。

各サービスは次のことを行います:

RabbitMQ を使用して特定のイベントをリッスンします。
新しいイベントを発行して後続のステップをトリガーします。

Docker を使用した RabbitMQ のセットアップ

イベントキューとして RabbitMQ を使用します。 Docker を使用してローカルで実行します:

docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

ログイン後にコピー

http://localhost:15672 で RabbitMQ 管理インターフェイスにアクセスします (ユーザー名: guest、パスワード: guest)。

交換、キュー、バインディングのセットアップ

イベントに対応できるように RabbitMQ を構成する必要があります。 RabbitMQ インフラストラクチャをセットアップするための init.go ファイルの例を次に示します。

package main

import (
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    err = ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare an exchange: %v", err)
    }

    _, err = ch.QueueDeclare("PatientVerified", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    err = ch.QueueBind("PatientVerified", "PatientVerified", "events", false, nil)
    if err != nil {
        log.Fatalf("Failed to bind a queue: %v", err)
    }
}

ログイン後にコピー

完全なコードはここにあります!

注: 運用環境では、GitOps アプローチ (Terraform など) を使用してこの設定を管理するか、各サービスが独自のキューを動的に処理できるようにすることができます。

実装: サービスファイル

各サービスには独自の main.go があります。障害を適切に処理するための補償アクションも含まれます。

1.患者サービス

このサービスは患者の詳細を確認し、PatientVerified イベントを生成します。また、下流側で障害が発生した場合には患者に通知することで補償します。

docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

ログイン後にコピー

2.スケジューラーサービス

このサービスは、PatientVerified をリッスンし、ProcedureScheduled を発行します。ダウンストリーム障害が発生した場合、手順をキャンセルすることで補償します。

package main

import (
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    err = ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare an exchange: %v", err)
    }

    _, err = ch.QueueDeclare("PatientVerified", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    err = ch.QueueBind("PatientVerified", "PatientVerified", "events", false, nil)
    if err != nil {
        log.Fatalf("Failed to bind a queue: %v", err)
    }
}

ログイン後にコピー

追加サービス

上記と同じ構造に従って、Inventory Service と Billing Service の実装を含めます。各サービスは前のイベントをリッスンして次のイベントを発行し、障害に対する補償ロジックが確実に導入されるようにします。

完全なコード ここにあります!

ワークフローの実行

RabbitMQ を開始します:

// patient/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[PatientService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "ProcedureScheduleCancelled")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[PatientService] Processing event: ProcedureScheduleCancelled")
            if err := notifyProcedureScheduleCancellation(); err != nil {
                log.Fatalf("Failed to notify patient: %v", err)
            }
        }
    }()

    common.PublishEvent(ch, "events", "PatientVerified", "Patient details verified")
    fmt.Println("[PatientService] Event published: PatientVerified")

    select {}
}

func notifyProcedureScheduleCancellation() error {
    fmt.Println("Compensation: Notify patient of procedure cancellation.")
    return nil
}

ログイン後にコピー

各サービスを実行します:
別のターミナルを開いて次を実行します:

// scheduler/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[SchedulerService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "PatientVerified")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[SchedulerService] Processing event: PatientVerified")
            if err := scheduleProcedure(); err != nil {
                common.PublishEvent(ch, "events", "ProcedureScheduleFailed", "Failed to schedule procedure")
                fmt.Println("[SchedulerService] Compensation triggered: ProcedureScheduleFailed")
            } else {
                common.PublishEvent(ch, "events", "ProcedureScheduled", "Procedure scheduled successfully")
                fmt.Println("[SchedulerService] Event published: ProcedureScheduled")
            }
        }
    }()

    select {}
}

func scheduleProcedure() error {
    fmt.Println("Step 2: Scheduling procedure...")
    return nil // or simulate a failure
}

ログイン後にコピー

出力の観察:
各サービスはイベントを順番に処理し、ワークフローの進行状況を記録します。