Kafka で注文サービスと在庫サービスを接続する

注文用の HTTP サービスと在庫のコンシューマーサービスをそれぞれ作成し、Kafka で order.created イベントを渡します。注文リクエストを送信した後、在庫サービスの処理ログまで確認します。在庫サービスは HTTP リスナーなしで実行します。

事前準備

先に CLI のインストールを完了してください。CLI 0.4.0 は、型を定義したコンシューマーのひな形を提供します。Kafka ブローカーとトランスポートの設定は別途準備します。

メッセージの配信を確認するには、localhost:9092 の Kafka ブローカーと order.created トピックが必要です。この例では、隔離されたローカル環境の平文接続を使用します。在庫コンシューマーはイベントをログに記録し、実際の在庫数は変更しません。

1. 2 つのサービスを作成する

作業ディレクトリを作り、生成するファイルを先に確認します。

mkdir spine-msa
cd spine-msa
spine-cli new order-app --layout legacy --module example.com/order-app --dry-run
spine-cli new stock-app --layout legacy --module example.com/stock-app --dry-run

プレビューを確認してから 2 つのアプリを作成します。以降のコマンドは、別途指定がなければ spine-msa/ で実行します。

spine-cli new order-app --layout legacy --module example.com/order-app
spine-cli new stock-app --layout legacy --module example.com/stock-app

2. 在庫コンシューマーのひな形を生成する

stock-app に、注文イベントを受け取るコンシューマーを生成します。まずプレビューを確認します。

spine-cli generate consumer order --root stock-app --topic order.created --field event_id:string --field order_id:int64 --dry-run

次のコマンドはコンシューマーのファイルを作成し、stock-app/internal/application/application.go に order.Register(app) の呼び出しとエラー確認のコードを追加します。生成直後のハンドラーは未実装エラーを返すため、検証を通過するには手順 5 のコードに置き換える必要があります。

spine-cli generate consumer order --root stock-app --topic order.created --field event_id:string --field order_id:int64

ファイル構成

共有イベントと注文コントローラーを追加し、生成されたコンシューマーの実装とアプリ設定を置き換えると、次の構成になります。

text

spine-msa/
├── order-app/
│   ├── go.mod
├── go.sum
│   ├── main.go
│   ├── spine.toml
│   ├── internal/application/
│   │   ├── application.go
│   │   └── application_test.go
│   └── src/
│       ├── events/order_created.go
│       └── controller/order_controller.go
└── stock-app/
    ├── go.mod
├── go.sum
    ├── main.go
    ├── spine.toml
    ├── internal/application/
    │   ├── application.go
    │   └── application_test.go
    └── src/consumer/order/
        ├── module.go
        ├── order_handler.go
        └── order_message.go

3. 共有イベントを定義する

order-app/src/events/order_created.go を新規作成します。2 つのサービスは、このパッケージのイベント型を共用します。stock-app が order-app のイベントパッケージをインポートできるよう、手順 6 で go mod edit を使ってローカルモジュールを接続します。

// order-app/src/events/order_created.go
package events

import "time"

type OrderCreated struct {
    EventID string `json:"event_id"`
    OrderID int64 `json:"order_id"`
    At time.Time `json:"at"`
}

func (e OrderCreated) Name() string { return "order.created" }
func (e OrderCreated) OccurredAt() time.Time { return e.At }

4. 注文サービスでイベントを発行する

order-app/src/controller/order_controller.go を新規作成し、order-app/internal/application/application.go を以下のコードに置き換えます。ルートの main.go と spine.toml はそのまま使用します。

publish.Event は現在のリクエストにイベントを追加し、戻り値はありません。ハンドラーが成功し、レスポンスの準備が完了すると、設定した Kafka パブリッシャーがイベントを送信します。

// order-app/src/controller/order_controller.go
package controller

import (
    "context"
    "fmt"
    "time"

    "example.com/order-app/src/events"
    "github.com/NARUBROWN/spine/pkg/event/publish"
    "github.com/NARUBROWN/spine/pkg/httperr"
    "github.com/NARUBROWN/spine/pkg/httpx"
    "github.com/NARUBROWN/spine/pkg/path"
)

type OrderController struct{}

func NewOrderController() *OrderController { return &OrderController{} }

func (c *OrderController) Create(ctx context.Context, id path.Int) (httpx.Response[string], error) {
    if id.Value <= 0 {
        return httpx.Response[string]{}, httperr.BadRequest("order ID must be positive")
    }
    publish.Event(ctx, events.OrderCreated{
        EventID: fmt.Sprintf("order.created:%d", id.Value),
        OrderID: id.Value,
        At: time.Now().UTC(),
    })
    return httpx.Response[string]{
        Body: "accepted",
        Options: httpx.ResponseOptions{Status: 202},
    }, nil
}

// order-app/internal/application/application.go
package application

import (
    platform "github.com/NARUBROWN/spine-echo-platform"
    "time"

    "example.com/order-app/src/controller"
    "github.com/NARUBROWN/spine"
    "github.com/NARUBROWN/spine/pkg/boot"
)

func BuildApplication() (spine.App, boot.Options, error) {
    app := platform.New()
    app.Constructor(controller.NewOrderController)
    app.Route("POST", "/orders/:order_id", (*controller.OrderController).Create)

    return app, boot.Options{
        Address: ":8080",
        EnableGracefulShutdown: true,
        ShutdownTimeout: 10 * time.Second,
        HTTP: &boot.HTTPOptions{},
        Kafka: &boot.KafkaOptions{
            Brokers: []string{"localhost:9092"},
            AllowInsecureTransport: true,
            Write: &boot.KafkaWriteOptions{TopicPrefix: ""},
        },
    }, nil
}

5. 在庫サービスでイベントを処理する

次の 3 つのファイルを、それぞれ以下のコードに置き換えます。order_message.go は共有イベントの型エイリアスに変更し、order_handler.go には未実装エラーの代わりに検証、重複確認、ログ出力を実装します。生成された module.go はそのまま使用します。

stock-app/internal/application/application.go を丸ごと置き換えるコードには、order.Register(app) が含まれています。生成時に追加された登録コードを重ねて追加しないでください。

// stock-app/src/consumer/order/order_message.go
package order

import "example.com/order-app/src/events"

type OrderEvent = events.OrderCreated

// stock-app/src/consumer/order/order_handler.go
package order

import (
    "context"
    "fmt"
    "log"
    "sync"
)

type OrderConsumer struct {
    mu sync.Mutex
    processed map[string]struct{}
}

func NewOrderConsumer() *OrderConsumer {
    return &OrderConsumer{processed: make(map[string]struct{})}
}

func (c *OrderConsumer) Handle(ctx context.Context, eventName string, event OrderEvent) error {
    if event.EventID == "" || event.OrderID <= 0 {
        return fmt.Errorf("invalid order event")
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    if _, exists := c.processed[event.EventID]; exists {
        return nil
    }
    log.Printf("topic=%s order_id=%d", eventName, event.OrderID)
    c.processed[event.EventID] = struct{}{}
    return nil
}

// stock-app/internal/application/application.go
package application

import (
    platform "github.com/NARUBROWN/spine-echo-platform"
    "time"

    "example.com/stock-app/src/consumer/order"
    "github.com/NARUBROWN/spine"
    "github.com/NARUBROWN/spine/pkg/boot"
)

func BuildApplication() (spine.App, boot.Options, error) {
    app := platform.New()
    if err := order.Register(app); err != nil {
        return nil, boot.Options{}, err
    }
    return app, boot.Options{
        EnableGracefulShutdown: true,
        ShutdownTimeout: 10 * time.Second,
        Kafka: &boot.KafkaOptions{
            Brokers: []string{"localhost:9092"},
            AllowInsecureTransport: true,
            Read: &boot.KafkaReadOptions{GroupID: "stock-service"},
        },
    }, nil
}

6. ローカルモジュールを接続し、コードを検証する

2 つのサービスは、それぞれ独立した go.mod と spine.toml を使用します。次のコマンドは spine-msa/ から開始し、2 つのモジュールを接続して検査した後、同じ場所に戻ります。各アプリに生成された main.go は、BuildApplication、Validate、Run を順に実行します。

cd order-app
gofmt -w src internal/application
go mod tidy
go mod download all
cd ../stock-app
go mod edit -require=example.com/order-app@v0.0.0
go mod edit -replace=example.com/order-app=../order-app
gofmt -w src internal/application
go mod tidy
go mod download all
cd ..
spine-cli doctor --root order-app
spine-cli doctor --root stock-app
spine-cli contract check --root order-app
spine-cli contract check --root stock-app
spine-cli verify --root order-app
spine-cli verify --root stock-app

各サービスはローカルの平文接続を使用するため、INSECURE_TRANSPORT_ENABLED の警告が 1 件ずつ出ます。通常の検査は通過しますが、--warnings-as-errors を使用すると失敗します。

contract check は静的な契約を検査します。verify は静的な契約に加え、gofmt、go vet、Go テストを検査します。ブローカー接続と実際のメッセージ配信は、次の手順で確認します。

7. 実行してイベントの配信を確認する

Kafka が localhost:9092 で起動しており、order.created トピックが存在することを確認します。各ターミナルの作業ディレクトリを spine-msa/ に合わせます。

1 つ目のターミナルで在庫サービスを起動します。

spine-cli run --root stock-app

2 つ目のターミナルで注文サービスを起動します。

spine-cli run --root order-app

3 つ目のターミナルで注文作成リクエストを送信します。

curl -i -X POST http://localhost:8080/orders/12345

HTTP 202 と本文 accepted が返されることを確認します。続けて在庫サービスのターミナルで、topic=order.created order_id=12345 のログを確認します。HTTP レスポンスは注文リクエストの結果であり、在庫サービスのログはイベント消費の結果です。サービス間の配信を検証するには、両方を確認する必要があります。

本番環境に適用する前に確認する点

トランスポートのセキュリティとトピック名

AllowInsecureTransport: true は、隔離されたローカルの平文ブローカーにのみ使用します。本番環境ではこのオプションを削除し、TLS/SASL とブローカーアドレスを設定します。既定のトランスポートは TLS 1.2 以上を使用します。トランスポート設定を直接指定する場合は、読み取りと書き込みの両方を保護する必要があります。

TopicPrefix は、トピック名の先頭に文字列をそのまま付加します。発行設定に prod. を使用する場合、コンシューマーの登録トピックも prod.order.created に合わせる必要があります。この例では、両方ともプレフィックスなしの order.created を使用します。検査と実行で同じ設定を使用してください。

重複処理と失敗するメッセージ

Kafka はメッセージを少なくとも 1 回配信するため、同じイベントが再度届くことがあります。同じイベントを再送する際にも維持されるイベント ID と、永続的な重複処理の防止が必要です。この例の重複処理の防止は、同じプロセス内でのみ有効です。実際の在庫変更と永続的な重複処理の防止は、同じ DB トランザクションで実装する必要があります。

ConsumerRetry はリーダーとトランスポート層の再構成を調整するもので、業務ハンドラーの再試行回数は制限しません。失敗し続けるメッセージは、パーティションの処理を止める可能性があります。メッセージを自動的にスキップしたり、DLQ に移動したりはしません。

DB の変更とイベントの発行

DB コミットとブローカーへの発行は、単一のアトミックな操作ではありません。DB の変更とイベント配信の整合性を保つ必要がある場合は、transactional outbox とコンシューマーの重複処理の防止を別途実装してください。

次のステップ

セキュリティ設定で Kafka の接続設定を確認してください。実際の在庫リポジトリを接続する際は、イベント ID を永続的に記録し、在庫変更と一緒にコミットするよう構成してください。

Kafka で注文サービスと在庫サービスを接続する ​

事前準備 ​

1. 2 つのサービスを作成する ​

2. 在庫コンシューマーのひな形を生成する ​

ファイル構成 ​

3. 共有イベントを定義する ​

4. 注文サービスでイベントを発行する ​

5. 在庫サービスでイベントを処理する ​

6. ローカルモジュールを接続し、コードを検証する ​

7. 実行してイベントの配信を確認する ​

本番環境に適用する前に確認する点 ​

トランスポートのセキュリティとトピック名 ​

重複処理と失敗するメッセージ ​

DB の変更とイベントの発行 ​

次のステップ ​