Apache Iggy によるイベント

wakuwaku::iggy（wakuwaku-iggy クレート）は wakuwaku のイベント層です。イベント型を Rust の型として一度定義すれば、残りはフレームワークが引き受けます。どの Iggy トピックとパーティションに送るか、ボディをどうエンコードするか、どのイベントを順番どおりに処理すべきか、どれをまとめたり畳み込んだりできるか、ハンドラが失敗したらどうするか、といった点です。

Spring に慣れている方は、Spring for Apache Kafka が担っている役割、つまり型付きのパブリッシャーとリスナーメソッドを思い浮かべてください。違いは判断の置き場所です。wakuwaku では順序とバッチ処理のルールをイベント型に宣言し、コンパイラがそれを検査します。コンシューマはその宣言を使い、イベントが求める順序を崩さない範囲でパーティションを並行に処理します。

なぜパーティション単位ではなくキー単位で順序を守るのかリンクをコピー

Iggy のパーティションは全順序のログです。最も単純なコンシューマはレコードを厳密に 1 件ずつ処理しますが、これには 2 つのコストがあります。

• 1 つのキーが他のキーを止める。 1 つのパーティションには多くのキー（ユーザー、注文など）のイベントが混在します。注文 17 のイベント処理が遅かったり失敗し続けたりすると、同じパーティションにある他の注文はすべてその後ろで待たされます。
• 何もまとめられない。 あるユーザーへの「ポイント付与」イベントが 10 件続けば、合計を 1 回渡せば済む場合でもハンドラ呼び出しは 10 回になります。

業務ロジックがパーティション全体の順序を必要とすることはほとんどありません。必要なのは1 つのキーに属するイベントの順序です。注文は支払いより先に作成されていなければなりませんが、注文 17 と注文 18 は互いに無関係です。そこで wakuwaku はキー単位で順序を守ります。

• 異なるキーのイベントは、同じパーティションにあっても互いに順序づけられません。キー同士は並行に処理されます。
• 1 つのキーの中では、同じ型のイベントは常にログ順を保ちます。
• 1 つのキーの中の異なる型のイベント同士については、各イベントの EventAtomicOrdering（Relaxed、Acquire、Release、AcqRel）が、何が何を追い越してよいかを決めます。詳しくは 順序 を参照してください。
• ポーリングした 1 バッチの中で、2 つの順序フェンスの間にある同じ型・同じキーのイベントすべてが ラン (run) になります。イベント型の代数（NonAssociative、Commutative、Idempotent、IdempotentCommutative、Associative）は、ハンドラに渡す前にランをどう縮約してよいかを表します。詳しくは 代数 を参照してください。

順序がキーの範囲に閉じているので、失敗したキーだけを脇へ退かすこともできます。そのキーの未完了の処理はリトライトピックへ移り、パーティションの残りはそのまま流れ続けます。詳しくは エラー処理 を参照してください。

モデルリンクをコピー

イベントの送り先と識別方法は、次の 3 つの型で決まります。

ストリームは型には含まれず、設定から決まります。Publisher::new と ConsumerConfig に渡すストリーム名です。キー型ごとに 1 つのトピックなので、注文に関するすべてのイベント（作成、支払い、発送など）は OrderId をキーとする 1 つのトピックに集まります。そのため 1 つのコンシューマがそれらすべてをログ順に受け取り、イベント間の順序を守れます。

コンシューマが行うことリンクをコピー

IggyConsumerRegisterCenter::start は、設定したパーティションごとに 1 つのタスクを起動します。各タスクは次の段階を繰り返します。

1. まずリトライ。 リトライトピック retry_<topic> の自分のパーティションを読み、リトライの待ち時間が過ぎたキーを再実行します。
2. ポーリング。 メイントピックの自分のパーティションから最大 poll_count 件のレコードを取得します。
3. 計画。 バッチをキーのハッシュでグループ化し、各キーのイベントを順序フェンスでセグメントに分け、各セグメントのイベントを型ごとにランにまとめます。
4. 実行。 キー同士は並行に処理します。1 つのキーの中ではセグメントを順に処理し、セグメントの中では先頭 (head) のラン、本体 (body) のラン群を並行に、最後に末尾 (tail) のランを処理します。
5. 縮約してハンドラへ。 各ランをデコードし、イベント型の代数で縮約し、最大 MAX_BATCH 件ずつのチャンクに分けてハンドラを呼び出します。
6. 失敗の隔離。 リトライ可能な失敗が起きると、そのキーの未完了の処理をリトライトピックへ移し、キーを隔離します。以後のそのキーのイベントはリトライトピックでその後ろに並びます。
7. コミット。 DeliveryMode に応じて、処理の前または後にコンシューマのオフセットを保存します。

API とすべての設定項目は イベントの購読 で順に説明します。

フィーチャーを有効にするリンクをコピー

イベント機能は iggy フィーチャーの後ろにあり、デフォルトでは無効です。

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["iggy"] }
tokio = { version = "1", features = ["full"] }

このフィーチャーで次のものが使えるようになります。

• wakuwaku::iggy: 再エクスポートされた wakuwaku-iggy クレート（イベント、パブリッシャー、ハンドラ、コンシューマのランタイム）。
• wakuwaku::deps::iggy: wakuwaku がビルドに使っている Apache Iggy のクライアント SDK（iggy 0.11）。IggyClient やストリーム・トピックの作成にはこれを使ってください。iggy を自分で依存に加えるより、バージョンが常に一致します。
• ServiceBuilder::iggy_consumer と ServiceBuilder::compose_iggy_consumer: サービスの共有依存からイベントハンドラを組み立てる機能。

イベントのボディは kanau::message のトレイトでシリアライズします。serde の JSON で derive するには依存を 2 つ追加する必要があります。ボディのシリアライズ を参照してください。

このセクションの内容リンクをコピー