状態変更を、疎結合に伝える — Kafka(MSK)によるイベントハブ連携

在庫・債権・通知…。ひとつの状態変更に多くの後続が反応する。それをイベントで疎結合につなぐ。

WHY状態変更に、後続が群がる

通販基幹では、ひとつの状態変更が多くの後続処理を呼び起こす。受注が確定すれば、在庫を引き当て、債権を計上し、顧客へ通知し、外部システムへデータ連携する。これらを受注処理から同期的に呼び出すと、呼び出し先が増えるほど処理は肥大し、どれかひとつが遅延・失敗しただけで受注そのものが巻き込まれてしまう。発行する側が「誰が反応するか」を知っている限り、結合は解けない。

そこで、状態変更をドメインイベントとして表現し、マネージドの Kafka(MSK)へ発行する。発行側は「何が起きたか」を宣言するだけで、誰が反応するかは知らない。消費側は各自の関心に応じてトピックを購読する。発行と消費が時間的にも実装的にも切り離される。

FLOW1つの発行、複数の反応

発行側(プロデューサ)はイベントをトピックへ1度書き込むだけでよい。在庫・債権・通知といった消費側(コンシューマ)は、それぞれ独立してトピックを読み、自分の処理を進める。片方が止まっていても、もう片方の処理は影響を受けない。

GATE入口を、ひとつに統一する

Vegas の中央ゲートウェイは、同期呼び出しやキューだけでなく Kafka レコードも入口として受け取る。届いたレコードをデコードして正規化し、以降は同一の処理系に載せる。呼び出し経路(同期/非同期/イベント)が違っても、業務ロジックは1本にまとまる。

• 同期:即時応答が必要な要求。
• 非同期(キュー):後回しでよい単発の処理。
• イベント(Kafka):1対多で広く波及する状態変更。

CODEレコードを受けて処理に載せる

Kafka レコードの値は base64 でエンコードされて届く。ゲートウェイはこれをデコードしてイベントとして復元し、既存の処理系へ引き渡す。

def handle_kafka(records):
    for topic, msgs in records.items():
        for m in msgs:
            # 値は base64。デコードしてイベントへ復元
            raw = base64.b64decode(m["value"])
            event = decode_event(raw)

            # 冪等性:同一キーの再処理は握りつぶす
            if already_processed(event.dedup_key):
                continue

            # 同期/キューと同じ処理系に載せる
            dispatch(event)
            mark_processed(event.dedup_key)

TRADEOFF疎結合の代償を、設計で払う

疎結合は多くを与えるが、無償ではない。得られるのは、発行側と消費側の独立、消費側を後から追加できる拡張性、スパイクをトピックが吸収する耐性、そしてオフセットを戻して再処理できる回復性である。一方で、次の代償を設計で払う必要がある。

• 順序保証:順序が要る事象は、同一エンティティを同じパーティションへ寄せるキー設計が要る。
• 重複配信:少なくとも1回の配信が前提。消費側は冪等でなければならない(上記の dedup_key)。
• 観測・トレース:非同期で経路が分岐するため、相関IDでイベントを追える仕組みが要る。
• スキーマ進化:発行側と消費側が別々に変わる。互換性を保つスキーマ管理が要る。

要は、同期の「その場で分かる」明快さを手放す代わりに、拡張性と耐障害性を得る取引だ。即時整合が絶対の処理まで無理にイベント化しないこと、逆に波及の広い状態変更はイベントに寄せること。その線引きが設計の勘所になる。