メッセージングと pub/sub — プロセスをまたいでイベントを届ける

第 12 章のドメインイベントは、同じアプリケーションの中で発行され、同じアプリケーションの中のリスナーが受け取ります。キューに載せて実行を後回しにしても、送り手と受け手が同じコードベースにいることは変わりません。

受け手が別のアプリケーションになると、あいだに立つものが要ります。本章はその仲介役を扱います。ブローカーが何を引き受けているかが分かれば、キューか pub/sub かという選択も、順序や重複の保証がどこまで効くかも、そこから導けます。

プロセスの中のイベントと、外へ出すイベント

プロセスの中で完結するイベントの仕組みはデザインパターンガイドの Observer が扱っています。Laravel のイベントとリスナーもその形です。

外へ出すときに増えるものが 3 つあります。

送り手と受け手が別のプロセスにいます。同じメモリを共有しないので、渡せるのは直列化した値だけです。オブジェクトの参照も、開いているトランザクションも渡りません。

受け手がその瞬間に動いているとは限りません。プロセス内なら呼び出した先は必ず存在しますが、外へ出すと相手が停止中のこともあります。届けるのを諦めるか、相手が戻るまで預かるかを決めなければなりません。

届いたかどうかが分かりません。メソッド呼び出しなら戻り値か例外で結果が分かります。ネットワークを挟むと、送った側は「届かなかった」と「届いたが返事が返らなかった」を区別できません。

この 3 つを引き受けるのがメッセージブローカーです。

ブローカーが引き受けているもの

第 12 章にも 1 つのイベントが複数のリスナーへ配られる図がありますが、そちらはプロセスの中の話でした。ここで違うのは、受け手のまとまりごとに進み具合が独立していることです。在庫サービスがどこまで処理したかと、会計サービスがどこまで処理したかは別々に記録されます。片方が止まってももう片方は進みます。

ブローカーが引き受けているものを 4 つに分けます。

保持 — 送り手と受け手の時間を切り離す

送り手が投げた時点で受け手が動いている必要をなくすのが、保持の役目です。受け手が落ちていても、戻ってきたときに続きから読めます。

いつ消えるかはブローカーの型で変わります。受け取った時点で消える型と、受け取っても残る型があります。Kafka は後者で、消える条件は消費ではなく保持期間です。公式ドキュメントは「従来のメッセージングシステムと違い、イベントは消費後に削除されない。代わりに、トピックごとの設定でどれだけ保持するかを決め、古いイベントはその後に破棄される」と書いています¹。

この違いが効くのは、受け手を後から足すときです。残る型なら、新しい受け手が過去のイベントを最初から読み直せます。消える型では、その受け手が参加する前のイベントはもう手に入らないので、受け手を足すたびに送り手へ再送を頼むことになります。

分配 — 1 人が取るか、全員に配るか

同じメッセージを誰が受け取るかには 2 通りあります。

キューは 1 件を 1 人が取ります。10 件のメッセージにワーカーが 2 つぶら下がれば、それぞれが 5 件ずつ処理します。処理を分担させたいときの形です。

pub/sub は 1 件を全員に配ります。注文が確定したことを在庫サービスと会計サービスの両方に知らせたいなら、こちらになります。

実際のブローカーは、この 2 つを組み合わせて表現します。Kafka は Consumer Group という単位を持ち、グループが違えば同じイベントが両方に届き、同じグループの中では分担になります。公式ドキュメントは「購読しているトピックの各メッセージを、各 Consumer Group の 1 プロセスへ届ける。これはグループの全メンバーでパーティションを分け合い、各パーティションがグループ内のちょうど 1 つのコンシューマーに割り当てられることで実現される」と説明しています²。

RabbitMQ は分配の判断を exchange という部品に持たせます。exchange がメッセージを受け取り、ルーティングの規則に従って 0 個以上のキューへ振り分けます³。全員に配りたいなら fanout 型の exchange を使い、こちらはルーティングキーを無視してつながっているキューすべてへ流します。

再配送 — 受け取ったことをどう確かめるか

受け手が読んだ直後に落ちたら、そのメッセージは処理されていません。ブローカーはそれを検出して配り直します。

判断の材料は受け手からの応答です。RabbitMQ のドキュメントは「メッセージの確認応答を使っている場合、ブローカーがキューからメッセージを完全に取り除くのは、そのメッセージ（またはメッセージ群）に対する通知を受け取ったときだけ」と書いています⁴。応答が来なければ残り続け、別の受け手に渡ります。

再配送があるということは、同じメッセージが 2 回届くということです。受け手は同じ処理を 2 回されても結果が変わらない形で書く必要があります。作り方は第 12 章がリスナーの側で扱っており、一般的な考え方はウェブエンジニア基礎知識ガイドにあります。

順序 — どの単位で保たれるか

「順に届く」という保証は、ブローカー全体では成り立ちません。成り立つのは、もっと小さい単位の中だけです。

Kafka の単位はパーティションです。公式ドキュメントは「同じイベントキー（顧客 ID や車両 ID など）を持つイベントは同じパーティションに書かれ、あるトピックパーティションのコンシューマーは、書かれたのとまったく同じ順序でそのパーティションのイベントを読むことを Kafka が保証する」と書いています⁵。キーを注文 ID にすれば、1 つの注文に起きたことは順に届きます。別々の注文どうしの前後は決まりません。

Amazon SQS の FIFO キューはメッセージグループが単位です。ドキュメントは「同じメッセージグループのメッセージは常に 1 件ずつ、厳密な順序で処理される」とし、続けて「標準キューは MessageGroupId を使わず、順序の保証を提供しない。厳密な順序が要るなら FIFO キューを使うこと」と書いています⁶。

順序の単位を決めるのは設計です。注文 ID を単位にすれば 1 つの注文は順に届きますが、単位の中は 1 件ずつしか処理できないので、単位を細かくするほど並列に処理できます。順序を強く取るほど遅くなる、という交換になります。

3 つのブローカーで型が違う

代表的な 3 つを並べます。行は保持と分配と順序に、消えるときから決まる「過去を読み直せるか」を足した 4 つです。再配送は型ごとの違いを確かめていないので表に出しません。

RabbitMQ にはキューの種類がいくつかあり、ここまでの説明は確認応答で消える型 (classic / quorum) を指しています。ストリームという種類だけ性質が違うので、列を分けます。

	Kafka	RabbitMQ (classic / quorum)	RabbitMQ (stream)	Amazon SQS
消えるとき	保持期間が過ぎたとき¹	確認応答を受け取ったとき⁴	保持期間が過ぎたとき⁷	受け手が削除したとき、または保持期間が過ぎたとき⁸
分配の表現	Consumer Group²	exchange の型³	exchange の型³	標準キューと FIFO キュー⁶
順序の単位	パーティション（キー単位）⁵	—	—	FIFO はメッセージグループ、標準は保証なし⁶
過去の読み直し	できる¹	できない⁴	できる⁷	できない⁸

RabbitMQ の順序を 2 列とも空欄にしたのは、引いたドキュメントに保証の記述が無かったためです。確認していないので、ここに「保証される」とも「されない」とも書きません。採用するなら、使う版のドキュメントで確かめてください。

型を選ぶ分かれ目は、過去を読み直したいかどうかに出ます。読みモデルを何度でも作り直したい、受け手を後から足したい、という要求があるなら残る型を選びます。処理を分担させたいだけなら、消える型のほうが運用は軽くなります。

配送保証は at-least-once から考える

配送の保証には 3 つの言い方があります。

at-most-once — 重複しないが、失われることがある
at-least-once — 失われないが、重複することがある
exactly-once — 失われず、重複もしない

実務で既定になるのは at-least-once です。再配送を持つブローカーは、応答が返らなかったメッセージを配り直すので、必然的に重複が起きます。

exactly-once は、条件を限れば成り立ちます。SQS の FIFO キューは重複排除の識別子を持ち、「同じ重複排除 ID のメッセージが 5 分の重複排除期間内に複数送られた場合、それらは重複として扱われ、1 つだけが配信される」と定めています⁹。期間が区切られていることに注意してください。5 分を超えて届いた再送は別のメッセージになります。

設計としては、ブローカーの保証に頼らず受け手を冪等に書くほうが安く済みます。重複排除の仕組みは期間や識別子の付け方に前提があり、その前提が崩れる条件を全部追うより、2 回来ても壊れない処理を書くほうが確実です。

Outbox パターン — 保存と送信を 1 つのトランザクションに入れる

集約を保存してブローカーへ送る、という処理には穴があります。保存は成功したのに送信が失敗すると、外の世界は注文が確定したことを知らないまま残ります。逆に、送信は成功したのに保存がロールバックすると、起きていないことが外へ広まります。

後者はLaravel API 開発ガイドのキューの章が扱っています。トランザクションの中でキューへ投入すると、ワーカーがまだコミットされていない注文を読みに行き、ロールバックしても投入したぶんだけが残る、という形です。本章はその続きで、名前の付いた対処を扱います。

本ガイドは第 19 章で DB::afterCommit を使う形を採っています。コミットを待ってから配信するので、ロールバックした注文のイベントは出ていきません。

それでも残る穴があります。コミットの後、配信の前にプロセスが落ちると、そのイベントは永久に出ていきません。注文は確定しているのに、在庫サービスも会計サービスも知らない状態が残ります。しかも例外もログも出ません。誰かが注文一覧と在庫の突き合わせに気付くまで表に出ない型の壊れ方です。

Outbox パターンは、送るべきイベントを集約と同じトランザクションで自分のデータベースへ書きます。

Schema::create('outbox_messages', function (Blueprint $table) {
    $table->bigIncrements('id');
    // 送り先のトピック名。受け手は別のアプリケーションなので、PHP のクラス名でなく
    // ブローカー側で通じる名前を入れる
    $table->string('topic');
    // 順序の単位を決める値。ブローカーへ渡すキーで、注文 ID が入る。
    // payload の中から掘らずに列で持つ (「順序 — どの単位で保たれるか」の節)
    $table->string('partition_key');
    $table->json('payload');
    // 送信済みになった時刻。null なら未送信
    $table->timestamp('published_at')->nullable();
    $table->timestamp('created_at');

    // 送信プロセスが引くのは published_at IS NULL の行を id 順に、という形だけ。
    // 左端が published_at なのはそのため
    $table->index(['published_at', 'id']);
});

書き込みは注文の保存と同じトランザクションに入ります。第 17 章の save() に足す形になります。

// app/Infrastructure/Repository/EloquentOrderRepository.php

// イベントのクラス名をブローカー側の名前へ写す。受け手が別のアプリケーションなので、
// PHP の名前空間をトピック名に漏らさない。イベントを足してここを更新し忘れると
// 添字の解決に失敗し、保存そのものが落ちる
private const TOPICS = [
    OrderConfirmed::class => 'order.confirmed',
    OrderCancelled::class => 'order.cancelled',
    OrderShipped::class => 'order.shipped',
];

public function save(Order $order): void
{
    // pullDomainEvents() は取り出してクリアする (第 12 章) ので、呼ぶのは 1 回だけ。
    // outbox への書き込みとプロセス内への配信が、この同じ配列を使う
    $events = $order->pullDomainEvents();

    DB::transaction(function () use ($order, $events) {
        // 集約の保存は第 17 章のまま
        $orderModel = OrderModel::findOrNew($order->id()->value());
        $orderModel->id = $order->id()->value();
        $orderModel->fill([/* 第 17 章と同じ */])->save();
        $this->saveOrderLines($orderModel, $order->orderLines());

        // 外へ出すイベントを同じトランザクションで積む。ここが要点で、
        // 注文が保存されたなら送信予定も必ず残り、片方だけということが起きない
        DB::table('outbox_messages')->insert(
            array_map(fn (object $event): array => [
                'topic' => self::TOPICS[$event::class],
                // 同じ注文のイベントを同じパーティションへ寄せるためのキー
                'partition_key' => (string) $order->id()->value(),
                'payload' => json_encode($this->toPayload($event), JSON_THROW_ON_ERROR),
                'created_at' => now(),
            ], $events),
        );
    });

    // プロセス内のリスナーへの配信は第 19 章のまま。outbox は外向けの経路を
    // 足すだけで、こちらを置き換えない
    DB::afterCommit(fn () => $this->eventDispatcher->dispatchAll($events));
}

toPayload() は、イベントを外へ出せる連想配列に直す処理です。第 12 章のイベントは @return object[] として扱われる素のオブジェクトで、直列化の口を持ちません。

この変換はリポジトリ側に置きます。第 23 章は同じ必要から StorableEvent というインターフェースをドメイン層に足していますが、あちらが作るのは内部の正本の形で、外へ出す形とは別の決めごとです。イベント側に口を開けると、ブローカーの都合がドメインに入ります。

なお $events を取り出す位置が第 17 章と違います。あちらはトランザクションの後で pullDomainEvents() を呼びますが、outbox への insert がトランザクションの中でこの配列を要るので、前へ動かしました。ロールバックしたときは outbox の行も残らず、配信も起きません。ただし $order インスタンスからはイベントが失われているので、やり直すなら集約を読み直します (第 19 章)。

送信は別のプロセスが担います。published_at が null の行を読み、ブローカーへ渡し、成功したらその列を埋めます。プロセスが落ちても行は残り、次の起動で拾われます。

この形でも重複は消えません。ブローカーへ渡した直後、published_at を埋める前に落ちれば同じメッセージが 2 回出ます。at-least-once であることは変わらず、受け手が冪等であることは引き続き要ります。Outbox が保証するのは「送り漏らさない」ことだけです。

Polling Publisher と Transaction Log Tailing

送信プロセスの作り方には 2 通りあります。

Polling Publisher は、未送信の行を定期的に問い合わせます。実装が素朴で、テーブルを見れば何が滞っているかも分かります。代わりに、問い合わせの間隔がそのまま配信の遅れになり、間隔を詰めると空振りのクエリが増えます。

Transaction Log Tailing は、データベースのレプリケーション用のログを読んで追記を検出します。ポーリングの負荷も遅れも無くなりますが、データベースごとのログ形式を読む部品が要ります。

小さく始めるなら Polling Publisher です。上の published_at を使う形がそれで、送信済みかどうかを行そのものが持ちます。「どこまで読んだか」を位置で覚えないので、第 23 章の通し番号の穴を踏みません。後からコミットされた行も、published_at が null のまま次の周回で拾われます。

送信済みの印を行に持てない送信元を選ぶと、この前提が崩れます。次の節がそれに当たります。

イベントソーシングを採っているなら

第 23 章のイベントストアを持っているなら、outbox のテーブルを省ける見込みがあります。order_events が既に「起きたことを追記順に並べたテーブル」で、集約の保存と同じトランザクションに入っているためです。送信プロセスはそこを読みます。

役割は別だということは、先に押さえておきます。イベントストアは内部の正本で、そこから集約を再構成します。外へ出すのは通知であって正本ではありません。1 つのテーブルを 2 つの目的に使うので、外向けに出すイベントの形を内部の都合で変えると受け手が壊れます。外へ出すぶんだけ別の形に変換する層を挟むのが安全です。

そして order_events には送信済みを示す列がありません。追記専用のテーブルなので後から印を付けられず、送信プロセスは「どの id まで送ったか」を別に記録することになります。ここで第 23 章のプロジェクションと同じ形になり、そちらで扱った通し番号の穴もそのまま当てはまります。id の続きから送る素朴な実装は、採番順とコミット順が一致しないために取りこぼし、そのイベントは二度と外へ出ません。

テーブルを 1 つ減らした代わりに、この穴への対処を引き受けることになります。第 23 章が挙げた 3 つがそのまま選択肢で、そこで「通し番号を自前で採る」を採るなら追記の側を直すことになるので、イベントストアの設計に戻ります。どれも重いと判断するなら、イベントソーシングを採っていても outbox のテーブルを別に持ちます。同じ内容が 2 か所に書かれる代わりに、送信済みの印を行に持てます。

イベントソーシングを採っていないなら、選択肢は outbox のテーブルだけです。

なお order_events を送信元にするなら、パーティションキーは outbox の列でなく第 23 章の stream_id を使います。集約 1 件のストリームを指す値なので、同じ集約のイベントが同じパーティションへ寄ります。

メッセージに何を載せるか

toPayload() が作る中身をどこまで詰めるかは、それ自体が設計の判断です。Martin Fowler は 2017 年の整理で、イベントを使う型を 4 つに分けています¹⁰。このうち外へ出すイベントの中身に関わるのが、最初の 2 つです。残る Event Sourcing と CQRS は第 23 章が扱います。

型	載せるもの	受け手がすること
Event Notification	起きたことと識別子。同記事は「多くのデータを載せる必要はなく、しばしば識別子と、詳しく問い合わせるための送り元へのリンクだけ」と書いています	足りない情報を送り元に問い合わせる
Event-Carried State Transfer	同記事の言葉で「変わったデータの詳細」	自分の複製を更新し、送り元に問い合わせない

分かれ目は、受け手が送り元に問い合わせてよいかどうかです。

Event Notification は載せる情報が少ないぶん、送り元が内部の形を変えても受け手が壊れにくくなります。代わりに受け手は毎回問い合わせるので、送り元が落ちていれば仕事が止まります。同記事はこの型について、システムをまたぐ大きな流れが見えにくくなる点も挙げています。

Event-Carried State Transfer は逆です。受け手が自分の複製を持つので、送り元が落ちていても動けます。代わりに、複製を持つということはその複製が古い瞬間があるということで、受け手の側に「いつの時点のデータか」を扱う仕事が増えます。

注文の例で言えば、order.confirmed に注文 ID だけを載せるのが前者、配送先や明細まで載せるのが後者です。受け手が別のチームや別の会社なら、問い合わせの口を開け続ける約束のほうが重いので、後者に寄ります。

外へ出すイベントの形をあとから変えるときの後方互換は、この章では扱いません (「この章が扱わないこと」を参照)。

Laravel で組むなら

Laravel のキューはブローカーではありません。公式ドキュメントが queue を「キューに積まれたジョブの山」と説明するとおり¹¹、単位はジョブで、1 つのジョブを 1 つのワーカーが処理します。この章で見た分配の型でいうとキューだけがあり、pub/sub に当たるものは無いというのが、ブローカーでないという意味です。

sqs ドライバを使っている場合でも同じです。SQS を経由してはいますが、Laravel が組み立てているのは自分のアプリケーションのジョブを自分のワーカーが拾う経路で、外のサービスが読む pub/sub にはなっていません。

別のアプリケーションへ届けるなら、ブローカーのクライアントを直接使うことになります。Outbox の送信プロセスを Artisan コマンドとして書き、その中でブローカーの SDK を呼ぶ形が素直です。送信プロセス自体をキューのジョブにすると、そのジョブの信頼性が今度は問題になるので、常駐かスケジュール実行にします。

なお config/queue.php の after_commit は、自分のアプリケーションのジョブをコミット後まで遅らせる設定です。公式ドキュメントは「開いている親トランザクションがコミットされるまで、Laravel は実際の dispatch を待つ」とし、「トランザクションの中で起きた例外でロールバックされた場合、そのトランザクション中に dispatch されたジョブは破棄される」と定めています¹²。これは自分のワーカーが拾うジョブの話で、外部への送信を保証するものではないので、Outbox の代わりにはなりません。

送信側 — outbox から Kafka へ

PHP から Kafka を扱うのは rdkafka 拡張です。C ライブラリの librdkafka を包んだ PECL 拡張で、RdKafka\Producer と RdKafka\KafkaConsumer を持ちます。

まず、外へ出すという操作をアプリケーション層のインターフェースにします。

// app/Application/Messaging/EventPublisherInterface.php
interface EventPublisherInterface
{
    /** @param array<string, mixed> $payload */
    public function publish(string $topic, string $key, array $payload): void;
}

このインターフェースに Kafka という語が出てこないのが要点です。外へ出すことはアプリケーション層の関心ですが、どのブローカーへどの設定で送るかはインフラ層の判断です (第 14 章)。

実装がインフラ層に来ます。

// app/Infrastructure/Messaging/KafkaEventPublisher.php
final class KafkaEventPublisher implements EventPublisherInterface
{
    private readonly \RdKafka\Producer $producer;
    private readonly \RdKafka\TopicConf $topicConf;
    /** @var array<string, \RdKafka\ProducerTopic> */
    private array $topics = [];
    private ?int $lastDeliveryError = null;

    public function __construct(string $brokers)
    {
        $conf = new \RdKafka\Conf();

        // 1 件ごとの配送結果はここにしか来ない。登録しないと、恒久的に
        // 配送できなかったメッセージも flush() は正常終了として返す
        $conf->setDrMsgCb(function (\RdKafka\Producer $p, \RdKafka\Message $m): void {
            // 失敗だけを記録し、成功では消さない。Producer は使い回すので、
            // 前回の未配送の結果がこの flush に混ざって届くことがある。
            // 成功で上書きすると、その失敗が見えなくなる
            if ($m->err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
                $this->lastDeliveryError = $m->err;
            }
        });

        $this->producer = new \RdKafka\Producer($conf);
        $this->producer->addBrokers($brokers);

        // acks はトピック側の設定。Conf に書いても既定のトピック設定として
        // 適用されるが、後から既定を差し替えると黙って消える。どのトピックに
        // 効かせるかを明示したいので TopicConf を作って渡す
        $this->topicConf = new \RdKafka\TopicConf();
        // 全レプリカへ書かれるまで成功と見なさない。緩めると、送ったつもりで
        // ブローカー側に残っていないイベントが出る
        $this->topicConf->set('request.required.acks', 'all');
    }

    public function publish(string $topic, string $key, array $payload): void
    {
        $this->lastDeliveryError = null;

        // 同じトピック名で 2 つ目を作っても、設定は最初に作ったものが使われる。
        // 作った handle を持ち回って、設定が黙って無視される形を避ける
        $this->topics[$topic] ??= $this->producer->newTopic($topic, $this->topicConf);

        // 第 1 引数はパーティション。RD_KAFKA_PARTITION_UA は「トピックの
        // パーティショナに任せる」で、第 4 引数のキーが同じものは同じ
        // パーティションへ行く。注文 ID をキーにすると 1 つの注文に
        // 起きたことが順に届く (「順序 — どの単位で保たれるか」の節)
        $this->topics[$topic]->produce(
            RD_KAFKA_PARTITION_UA,
            0,
            json_encode($payload, JSON_THROW_ON_ERROR),
            $key,
        );

        // produce() は内部キューに積むだけで送信は非同期。flush を挟まないと
        // プロセスが終わった時点で未送信ぶんが消える
        $flushed = $this->producer->flush(10_000);
        if ($flushed !== RD_KAFKA_RESP_ERR_NO_ERROR) {
            throw new \RuntimeException('Kafka への送信がタイムアウトしました');
        }

        // flush() が正常終了しても、それは「内部キューが空になった」だけ。
        // 配送できたかどうかは上のコールバックが受けた結果で判断する
        if ($this->lastDeliveryError !== null) {
            // エラーコードを載せる。載せないと、認証の失敗もトピック名の誤りも
            // サイズ超過も同じ 1 行になる。ただし前の行の未配送を拾うこともあるので、
            // 種別の手がかりであって行の特定には使えない
            throw new \RuntimeException(
                'Kafka がメッセージを受理しませんでした: ' . rd_kafka_err2str($this->lastDeliveryError),
            );
        }
    }
}

flush() の戻り値だけで成否を判断できないのが、ここでいちばん外しやすい点です。flush() が RD_KAFKA_RESP_ERR_NO_ERROR を返すのは「タイムアウトまでに内部キューが空になった」という意味で、空になった理由が恒久的な配送失敗でも同じ値になります。コールバックを登録しないと、送れていないのに published_at が埋まり、そのイベントは二度と外へ出ません。acks の設定が効くのも、結果を受け取る口があってこそです。

この実装は broker の接続先をコンストラクタで受けるので、コンテナに教える必要があります。第 17 章の $bindings も、同章が singleton 用に挙げている $singletons も、対応づけるのはインターフェースと実装クラスだけです。string $brokers のような素の値はどちらでも解決できないので、組み立て方を書ける register() を使います。

// config/services.php
'kafka' => [
    'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),
],

// app/Infrastructure/Provider/RepositoryServiceProvider.php (第 17 章のプロバイダ)
// 束縛は 1 か所に集めるので、ここに足す
public function register(): void
{
    // 接続を毎回張り直さないよう singleton にする
    $this->app->singleton(
        EventPublisherInterface::class,
        fn ($app) => new KafkaEventPublisher(config('services.kafka.brokers')),
    );
}

送信プロセスはこれを呼ぶだけです。

// app/Infrastructure/Messaging/PublishOutboxMessages.php (Artisan コマンド)
// outbox のテーブルを直接読むのでインフラ層に置き、アプリケーション層の
// EventPublisherInterface を呼ぶ
public function handle(EventPublisherInterface $publisher): void
{
    $rows = DB::table('outbox_messages')
        ->whereNull('published_at')
        ->orderBy('id')
        ->limit(100)
        ->get();

    foreach ($rows as $row) {
        try {
            $payload = json_decode($row->payload, true, 512, JSON_THROW_ON_ERROR);

            // キーは列から取る。payload の中を掘ると、イベントごとに形が違う
            // ぶんだけ送信側がイベントの中身を知ることになる
            $publisher->publish($row->topic, $row->partition_key, $payload);
        } catch (\Throwable $e) {
            // 1 行の失敗で残りを止めない。止めると、この行が先頭にいる限り
            // 次の周回も同じ場所で止まり、outbox 全体が進まなくなる
            Log::error('outbox の送信に失敗', ['id' => $row->id, 'error' => $e->getMessage()]);
            continue;
        }

        // 送信が確定してから印を付ける。ここで落ちれば次の周回で再送され、
        // 同じメッセージが 2 回出る。at-least-once はこの順序から来る
        DB::table('outbox_messages')
            ->where('id', $row->id)
            ->update(['published_at' => now()]);
    }
}

1 件ずつ flush するので速くはありません。まとめて produce してから 1 回だけ flush すれば速くなりますが、その場合はどこまで送れたかが分からなくなるので、published_at を一括で更新してよいかを別に確かめることになります。

continue で進めるようにした代わりに、2 つ引き受けたものがあります。

1 つは、失敗が静かになったことです。ブローカーが落ちていれば 100 行すべてが失敗し、ログが 100 行出るだけで送信自体は止まりません。published_at が埋まらないので送り漏らしはしませんが、気づく手がかりがログしかない状態です。運用では未送信のいちばん古い行がどれだけ滞留しているかを監視します。失敗した回数や最後のエラーを行に持たせる案もありますが、本章では列を増やしません。

もう 1 つは順序です。飛ばした行は、同じ注文の後続イベントより後に送られます。1 つの注文について「確定 → 出荷」の順に積んだうち確定の送信だけが失敗すると、次の周回で出荷の後に確定が届きます。パーティションキーが保証するのはブローカーへ渡した順であって、こちらが飛ばした順ではありません。受け手に順序への耐性まで求めないなら、失敗した行より後ろを送らずにその周回を打ち切る形にします。

このコマンドは 1 本だけ動かす前提です。published_at を埋めるのが送信の後なので、2 本が同時に回ると同じ行を両方が拾い、同じメッセージが 2 回出るだけでなく送られる順序も崩れます。パーティションキーで順序を揃えた意味が無くなるので、常駐 1 本にするか、スケジュール実行なら重複起動を止めるか、取り出しを lockForUpdate() で囲みます。

受信側 — Kafka から読みモデルへ

ここから先は受け取る側のアプリケーションの話です。送信側とは別のコードベースで、自分のイベントストアを持ちません。届いたメッセージから自分用のテーブルを作るだけなので、必要な契約も 1 つで足ります。

// 受け取る側の app/Application/Projection/OrderSummaryUpdaterInterface.php
interface OrderSummaryUpdaterInterface
{
    /** @param array<string, mixed> $payload */
    public function apply(string $topic, array $payload): void;
}

第 23 章の OrderSummaryProjectionInterface とは別物です。あちらはイベントのクラスごとに whenOrderPlaced() のようなメソッドを持ちますが、こちらが受け取るのはトピック名と連想配列だけで、送り元のクラスを知りません。知ってしまうと、送り元の内部の形に受け手が縛られます。

受け取る側の常駐プロセスがこれを呼びます。

// 受け取る側のアプリケーションの app/Infrastructure/Messaging/ConsumeOrderEvents.php
// (Artisan コマンド)。ブローカーを直接読むのでインフラ層に置き、アプリケーション層の
// OrderSummaryUpdaterInterface を呼ぶ
public function handle(OrderSummaryUpdaterInterface $updater): void
{
    $conf = new \RdKafka\Conf();
    $conf->set('metadata.broker.list', config('services.kafka.brokers')); // 受け取る側にも同じ設定を置く
    // Consumer Group。同じ group.id のプロセスどうしで分担し、違う group.id
    // には同じイベントが届く (「分配 — 1 人が取るか、全員に配るか」の節)
    $conf->set('group.id', 'order-summary-updater');
    // 読み切ったことを EOF で知らせてもらう。設定しないと下の EOF 分岐は来ない
    $conf->set('enable.partition.eof', 'true');
    // 初めて参加したときどこから読むか
    $conf->set('auto.offset.reset', 'earliest');
    // 自動コミットを切る。処理が終わってから自分でコミットする
    $conf->set('enable.auto.commit', 'false');

    $consumer = new \RdKafka\KafkaConsumer($conf);
    $consumer->subscribe(['order.confirmed', 'order.cancelled', 'order.shipped']);

    while (true) {
        $message = $consumer->consume(10_000);

        match ($message->err) {
            RD_KAFKA_RESP_ERR_NO_ERROR => $this->apply($updater, $consumer, $message),
            // 読み切った・待っただけ。どちらも異常ではないので回り続ける
            RD_KAFKA_RESP_ERR__PARTITION_EOF,
            RD_KAFKA_RESP_ERR__TIMED_OUT => null,
            default => throw new \RuntimeException($message->errstr(), $message->err),
        };
    }
}

private function apply(
    OrderSummaryUpdaterInterface $updater,
    \RdKafka\KafkaConsumer $consumer,
    \RdKafka\Message $message,
): void {
    $payload = json_decode($message->payload, true, 512, JSON_THROW_ON_ERROR);

    $updater->apply($message->topic_name, $payload);

    // 処理が終わってからコミットする。逆にすると、コミット後・処理前に
    // 落ちたメッセージが二度と来ない
    $consumer->commit($message);
}

コミットを処理の後に置くのが at-least-once です。落ちた位置によっては同じメッセージをもう一度受け取るので、apply() は 2 回呼ばれても結果が変わらない形で書きます。

1 件ごとに commit() するので、こちらも速くはありません。まとめてコミットすれば減らせますが、その分だけ落ちたときに再処理する範囲が広がります。while (true) を抜ける経路が無いので、止めるのはプロセスを落とすことになります。運用では停止信号を受けてループを抜け、close() を呼ぶ形にします。

apply() が投げるとループを抜けてプロセスが死にます。offset は未コミットなので、再起動すると同じメッセージをまた受け取り、同じ場所で落ち続けます。処理できないメッセージを脇へ避ける仕組み (デッドレターキュー) が要りますが、本章では扱いません (「この章が扱わないこと」を参照)。

もう 1 つ、RD_KAFKA_RESP_ERR__TIMED_OUT を黙って読み飛ばしていることにも注意してください。ブローカーに到達できない状態と、単に何も来ていない状態が、このコードでは区別できません。接続先の設定を間違えても「静かに動いているように見える」形になるので、setErrorCb で接続の失敗を拾うようにします。

ここまでが第 23 章の流れ図の ⑥ と ⑦ にあたります。受け取る側はイベントストアを持たず、届いたイベントから自分用のテーブルを組み立てるだけです。

拡張とラッパー

rdkafka は PECL 拡張なので、librdkafka を入れたうえで pecl install rdkafka で導入します。Laravel 向けのラッパーも公開されていますが、mateusjunges/laravel-kafka の現行版 (v2.11.5) が要求するのは illuminate ^12.0|^13.0 で、本ガイドが前提にする Laravel 11 では入りません。拡張を直接使えばこの制約を受けません。

入れる前に見積もるもの

ここまでが組み方の話です。実際に入れると決める前に、払うものを並べておきます。

払うもの	この章のどこで扱ったか
ブローカーそのものの知識	「ブローカーが引き受けているもの」「3 つのブローカーで型が違う」。保持・分配・再配送・順序の型は製品ごとに違い、選んだ製品の挙動を知らないまま設計できません
呼び出しの形が変わること	「プロセスの中のイベントと、外へ出すイベント」。戻り値も例外も返らない経路になり、結果が揃うのを待てなくなります
データベースとメッセージの整合性	「Outbox パターン」。保存はできたのに送信できていない状態を、自分で塞ぐことになります
メッセージの中身の決定	「メッセージに何を載せるか」。受け手との約束になるので、あとから変えるのが高くつきます
受け手を冪等にする作業	「配送保証は at-least-once から考える」。ブローカーの保証に関わらず必要です

5 つとも、ブローカーを入れた時点で発生します。このうち最初の 1 つだけがミドルウェアの話で、残りは自分のコードに降りてきます。

とくに 2 つ目は、書き方そのものが変わる代償です。メソッドを呼んで結果を受け取る形では書けなくなり、送ったあとのことは別の経路で知ることになります。ここを軽く見ると、非同期にしたのに結局どこかで待つ、という設計になります。

この転換を言語やランタイムの側が引き受ける設計もあります。アクターモデルがそれで、Akka の公式ドキュメントは「メソッドを呼ぶ代わりに、アクターは互いにメッセージを送る。メッセージの送信は、送り手から宛先へ実行スレッドを渡さない」と説明しています¹³。1 つのアクターが同時に処理するメッセージは 1 つなので、ロックを使わずに不変条件を保てる、という組み立てです。

ただしこれは並行処理のモデルの置き換えであって、この章の話の代わりにはなりません。アクターモデルはプロセスの中や同じクラスタの中でメッセージを扱う仕組みで、別のアプリケーションへ確実に届けるという問題は別に残ります。言語やランタイムに組み込みのアクターモデルを PHP は持たないので、本ガイドが採るのはブローカーを使う経路です。

この章が扱わないこと

ブローカーを実際に運用すると、ここから先に判断が要ります。本ガイドはドメインの設計を扱う本なので、そちらはサービス間連携を扱う別のガイドに譲ります。

論点	参照先
複数サービスにまたがる更新の整合性（Saga、補償トランザクション）	サービス間連携の設計ガイド
外へ出すイベントの形を変えるときの後方互換、デッドレターキュー	同上
ブローカーの運用（パーティション設計、監視、遅延の検出）	同上
プロセス内のイベントとリスナー	デザインパターンガイド、第 12 章
冪等な受け手の作り方	ウェブエンジニア基礎知識ガイド、第 12 章

まとめ

ポイント	内容
外へ出すと増えるもの	プロセスが別、受け手が動いているとは限らない、届いたか分からない
ブローカーの役目	保持、分配、再配送、順序の 4 つ
消えるときの違い	Kafka と RabbitMQ のストリームは保持期間、RabbitMQ の classic / quorum は確認応答、SQS は削除か保持期間。過去を読み直せるかがここで決まる
順序の単位	ブローカー全体では保証されない。Kafka はパーティション、SQS FIFO はメッセージグループ
配送保証	at-least-once が既定。受け手を冪等に書くほうが、重複排除の前提を追うより安い
Outbox	集約の保存と同じトランザクションに送信予定を書く。`DB::afterCommit` が埋めない「コミット後に落ちた」穴を埋める
メッセージの中身	識別子だけ載せて問い合わせさせるか、変わったデータごと載せるか。受け手が送り元に問い合わせてよいかで決まる
入れる代償	ブローカーの知識・呼び出しの形が変わること・DB との整合性・メッセージの中身の決定・受け手の冪等化の 5 つ。最初の 1 つ以外は自分のコードに降りてくる

参考資料

Apache Kafka「Introduction」 — 保持と順序の保証
RabbitMQ「AMQP 0-9-1 Model Explained」 — exchange の型と確認応答
Amazon SQS「FIFO queue and message identifiers」 — 順序の単位と重複排除
microservices.io「Transactional outbox」 — Outbox パターンの定式化
Laravel「Queues」 — ジョブとワーカーの関係、after_commit
Martin Fowler「What do you mean by "Event-Driven"?」 — Event Notification と Event-Carried State Transfer の区別
Akka「Introduction to Actors」 — アクターモデルの前提

次章では、これまで見た 4 層を組み合わせて、注文システムを実装します。

Apache Kafka「Introduction」。消える条件が消費でなく保持期間であることと、その期間をトピックごとに設定することを同ページが明記している。 ↩ ↩² ↩³
KafkaConsumer (Kafka 4.0 API)「Consumer Groups and Topic Subscriptions」。各パーティションがグループ内のちょうど 1 つのコンシューマーへ割り当てられること、新しいコンシューマーが加わると再割り当てが起きることを同ページが述べている。 ↩ ↩²
RabbitMQ「AMQP 0-9-1 Model Explained」。exchange が 0 個以上のキューへ振り分けること、その規則が exchange の型と binding で決まることを同ページが述べている。型は direct・fanout・topic・headers の 4 つで、fanout がルーティングキーを無視して結ばれた全キューへ流すことも同ページの記述による。 ↩ ↩² ↩³
同上。確認応答を使っているとき、キューから完全に取り除かれるのが通知を受け取った時点であることを同ページが述べている。 ↩ ↩² ↩³
Apache Kafka「Introduction」。同じキーのイベントが同じパーティションへ書かれること、パーティション内では書かれた順に読まれることを同ページが保証として述べている。 ↩ ↩²
Amazon SQS「FIFO queue and message identifiers」。同じメッセージグループが 1 件ずつ厳密な順序で処理されることと、標準キューが順序を保証しないことを同ページが述べている。 ↩ ↩² ↩³
RabbitMQ「Streams」。ストリームが追記専用のログで、期限が切れるまで繰り返し読めること、消費側が任意の位置から読み始められること、保持を時間 (max-age) と容量 (max-length-bytes) で決めることを同ページが述べている。 ↩ ↩²
Amazon SQS「DeleteMessage」。受け手が明示的に削除すること、保持期間を過ぎたメッセージを SQS が自動で削除することを同ページが述べている。 ↩ ↩²
Amazon SQS「FIFO queue and message identifiers」。5 分という重複排除期間はこのページの記述による。 ↩
Martin Fowler「What do you mean by "Event-Driven"?」 (2017)。Event Notification / Event-Carried State Transfer / Event Sourcing / CQRS の 4 つを区別している。 ↩
Laravel「Queues」の「Connections vs. Queues」節。1 つの接続が複数のキューを持ちうること、キューがジョブの山であることを述べている。 ↩
Laravel「Queues」の「Jobs & Database Transactions」節。 ↩
Akka「Introduction to Actors」。 ↩

プロセスの中のイベントと、外へ出すイベント​

ブローカーが引き受けているもの​

保持 — 送り手と受け手の時間を切り離す​

分配 — 1 人が取るか、全員に配るか​

再配送 — 受け取ったことをどう確かめるか​

順序 — どの単位で保たれるか​

3 つのブローカーで型が違う​

配送保証は at-least-once から考える​

Outbox パターン — 保存と送信を 1 つのトランザクションに入れる​

Polling Publisher と Transaction Log Tailing​

イベントソーシングを採っているなら​

メッセージに何を載せるか​

Laravel で組むなら​

送信側 — outbox から Kafka へ​

受信側 — Kafka から読みモデルへ​

入れる前に見積もるもの​

この章が扱わないこと​

まとめ​

参考資料​

Footnotes​