メッセージキュープロトコル

メッセージシステムは、バイトの交換方法だけでなく、メッセージの保存場所、いつ確認を行うか、そしてコンシューマーがどのように回復するかを定義します。RabbitMQ、Kafka、NATS を比較する際は、標準プロトコル、製品、および製品内のストレージモードを明確に区別する必要があります。「賢いブローカー、愚かなブローカー、最小限のブローカー」という分類だけでは、実際の障害挙動を十分に説明できません。

同じ製品内でも、ストレージタイプ、確認方式、レプリケーション設定の影響は、製品名間の差異よりも大きくなることがよくあります。RabbitMQ AMQP 1.0、NATS JetStream

「成功」が何を確定したのかをまず回答する

プロデューサーがバイトをソケットに渡したからといって、ブローカーが確実に受信したわけではありません。ブローカーが成功を返したからといって、サブスクライバーがビジネスデータベースを更新したわけではありません。コンシューマーが確認を行うかオフセットをコミットしても、ローカルトランザクションがコミットされたことを自動的に証明するものではありません。これらの境界は異なるプロセスで発生するため、明確な障害対策が必要です。

コンシューマー側で先に確認を行ってからビジネス処理を行うと、クラッシュ時に処理漏れが発生する可能性があります。先にビジネス処理を行ってから確認を行うと、確認の喪失により重複処理が発生する可能性があります。デジュープリケーションとローカルビジネス効果を同じトランザクション内に配置することで、この部分的なウィンドウをカバーできますが、外部のHTTP副作用に対しては、外部の冪等契約または補償に頼る必要があります。具体的な受信モデルについては、Webhook とコールバックを参照してください。

AMQP 0-9-1：キューへのルーティングとキューからの配信

パブリッシャーはメッセージを exchange に送信し、binding がどのキューに入るかを決定します。同じ fanout exchange に3つのキューがバインドされている場合、3つのキューはそれぞれ消費待ち状態を保持できます。同じキュー上の3つの競合コンシューマーは、通常、そのキューの配信を分担しているだけであり、各々がブロードキャストされたメッセージを1通ずつ受信しているわけではありません。

キューの durable、メッセージの persistent、レプリケーション、および publisher confirms は異なる設定です。永続キューであっても、すべてのメッセージが自動的に同じ方法でディスクに書き込まれるわけではありません。コンシューマーの ACK が、他の binding されたキュー内のコピーも消費されたことを意味するわけではありません。RabbitMQ Queues

パブリッシュ確認とルーティングされなかったメッセージ

RabbitMQ の publisher confirms は、AMQP 0-9-1 の拡張機能であり、パブリッシュ側の処理を確認するために使用されます。メッセージがどのキューにもルーティングされなかった場合でも、パブリッシュ確認が返されることがあります。この状況を検出するには、mandatory/return などのメカニズムを正しく使用する必要があります。パブリッシュ確認とコンシューマー ACK は完全に独立しています。RabbitMQ confirms

消費 delivery_tag のスコープは channel であり、それを受け取った同じ channel 上で確認を行う必要があります。再接続後、古い tag を永続メッセージIDとして扱うことはできません。multiple=true を指定すると、指定された tag までの複数の未確認配信が確認されますが、その間に完了していない作業がある場合、処理漏れのウィンドウが発生します。

手動確認モードでは、接続または channel の閉鎖により、通常、未確認の配信が再キューイングされます。自動確認モードでは、同様の保護は提供されません。requeue を行うと、観察される順序が変更される可能性があり、繰り返し nack/requeue を行うとホットサーキュレーションが発生するため、試行回数を記録し、一時的な失敗と処理不可能なメッセージに対して異なる戦略を採用する必要があります。

Prefetch は公平性のスイッチではない

prefetch は未確認の配信数を制限し、0 は通常、この上限を設定しないことを意味します。RabbitMQ では、非グローバル prefetch に対して、通常、コンシューマーごとに適用されます。2つのコンシューマーがそれぞれ prefetch=10 の場合、総上限を channel 全体で10と誤解してはいけません。Consumer Prefetch

1 に設定すると、コンシューマーあたりの在途メッセージ数が制限され、低速なコンシューマーが大量の仕事を先に占有するのを防げますが、スループットは確認の往復時間と処理時間にも影響を受けるため、各コンシューマーの処理負荷や所要時間が完全に公平になる保証はありません。非常に大きな値を設定すると、クライアントのメモリ増加、再送嵐、テールレイテンシの原因となる可能性があります。

バイトフレームとビジネスメッセージは階層化する必要がある

AMQP 0-9-1 フレームは、タイプ1 B、channel2 B、ペイロード長4 B、ペイロード、末尾の0xce で構成されます。1つのパブリッシュは、メソッドフレーム、コンテンツヘッダーフレーム、および複数のコンテンツボディフレームで構成される可能性があり、1回のソケット読み込みや1つのフレームを1つの完全なビジネスメッセージとして扱うことはできません。ハートビートは、タイプ8、channel0、空のペイロードを持つ8バイトのフレームです。RabbitMQ フレームフォーマット誤り訂正

import struct
heartbeat = struct.pack('!BHI', 8, 0, 0) + b'\xce'
assert heartbeat.hex() == '08000000000000ce'
assert len(heartbeat) == 8
print('AMQP 0-9-1 空ハートビートフレームの長さチェック通過')

AMQP 1.0：接続、セッション、およびリンクのフロー制御

AMQP 1.0 は、単に 0-9-1 のバージョン番号を変更しただけではありません。エンドポイントは connection 内で session を確立し、その後、単方向の link を attach します。メッセージは transfer によって運ばれ、delivery state と settlement によって結果が表現され、link credit が送信権限を制約します。すべての実装が完全に同一の exchange/queue 管理インターフェースを採用することを要求するものではありません。

同じ AMQP と呼ばれる2つのクライアントでも、バージョンが異なる場合は直接通信できません。移行時には、プロトコルバージョン、アドレスマッピング、認証、確認モード、および製品拡張を同時に確認する必要があります。「両方が AMQP をサポートしている」というだけでは、動作が一致するとは限りません。RabbitMQ AMQP 1.0 コンセプト

Kafka：ログ位置、グループの進捗、保持は異なる状態

Kafka はレコードをトピックのパーティションに追記し、各レコードにはそのパーティションのオフセットがあります。通常の consumer group での消費時には、各メンバーがパーティションを割り当てられ、レコードをフェッチします。異なるグループは、同じログを独立して読み取ることができます。グループのコミット位置は通常、Kafka によって管理・保存され、消費プロセス内だけに存在するわけではありません。

オフセットをコミットしても、対応するレコードが削除されるわけではありません。ログは、時間、容量、または圧縮などのポリシーに基づいて保持され、コンシューマーがデータがクリーンアップされた後に追いついた場合、元の位置では再再生できなくなる可能性があります。「コンシューマーがオフセットを自分で追跡する」というだけでは、ブローカーがグループ、トランザクション、または回復状態を管理していないと結論付けることはできません。Kafka Design

アプリケーションが 100〜103 を一度にフェッチして並列処理を行い、100 と 102 が先に完了した場合、安全な回復位置は依然として 101 です。101 も完了して初めて、103 に進めることができます。先に 103 をコミットすると、クラッシュ後に完了していない 101 をスキップしてしまいます。

def resume_offset(first, completed):
    result = first
    while result in completed:
        result += 1
    return result

assert resume_offset(100, {100, 102}) == 101
assert resume_offset(100, {100, 101, 102}) == 103
assert resume_offset(100, {101, 102}) == 100
# 連続したオフセットのみがこのセグメントに含まれるという明確な仮定；圧縮ログでは、実際の返却位置にギャップがある可能性がある。
print('連続した完了接頭辞と回復位置のチェック通過')

このモデルの連続した整数は、コミットルールを説明するためのものです。実際のログには、トランザクション、圧縮、またはその他の理由によるオフセットのギャップが含まれる可能性があるため、ある整数を受信できないからといって、無限に待機してはいけません。実際の返却レコードとパーティションの進捗に基づいて、処理済みの接頭辞を維持する必要があります。

非連続オフセット、バッチ境界、および再バランス後の古いタスク処理については、分散システムのシリーズ 消費グループと協調 を参照してください。本稿ではプロトコルレベルの最小限の例を残し、完全な回復進捗管理はそちらで展開します。

Kafka ワイヤープロトコルは固定された呼び出しパイプラインではない

リクエストは通常、長さプレフィックス、API key/version、correlation ID、および対応するバージョンのヘッダーを持ちます。レスポンスは correlation ID によってリクエストと関連付けられます。ApiVersions は能力のネゴシエーションに使用され、Metadata はノードとパーティションの情報を提供し、Produce/Fetch はレコードを処理し、グループプロトコルと OffsetCommit は消費の調整を管理します。プロデューサーは、各レコードをパブリッシュする前に、毎回 Fetch を実行するわけではありません。

リクエスト/レスポンスのスキーマは API バージョンによって変化し、新しい flexible versions ではコンパクトなエンコーディングと tagged fields が使用されます。1つの固定構造体ですべてのバージョンを解析することはできません。レコードはバッチ処理や圧縮される可能性があり、1つのリクエストが1つのメッセージに相当するわけではありません。Kafka Protocol

acks=all の確認は、ISR と最小同期レプリカなどの設定と組み合わせて理解する必要があり、すべての設定されたレプリカが常に同期済みであることを意味するわけではありません。冪等プロデューサーは、対応するプロデューサセッションとパーティションシーケンスを管理します。アプリケーションがプロデューサーを任意に再作成して同じビジネスイベントをコミットする場合、ビジネスレベルのデジュープリケーションを依然として考慮する必要があります。Kafka トランザクションは、Kafka の出力と消費位置を調整できますが、外部データベースを自動的にトランザクションに組み込むわけではありません。

102 が完了したのに、なぜ 101 しかコミットできないのか？

100 と 102 が先に完了し、安全な回復位置は 101 で停止します。誤って 103 をコミットすると、クラッシュ後に 101 をスキップしてしまいます。101 を補完した時点で、一度に 103 まで進めることができます。TCP の累積確認に似ているのは連続した接頭辞の制約ですが、ここでは「完了」がビジネス消費によって定義される点が異なります。

Core NATS：テキストコマンドとバイト区切りのペイロード

Core NATS は subject を介してオンラインの pub/sub を行い、queue group を使ってグループ内で配信を分担します。オフラインのサブスクライバーが逃した履歴メッセージをデフォルトで保存しません。PING/PONG は、すべてのサブスクライバーがビジネス処理を完了したことを意味するわけではありません。

クライアントプロトコルコマンドはテキストで表されますが、ペイロードはバイナリであり、長さフィールドは文字数ではなくバイト数です。SUB は subject、オプションの queue group、および sid を指定します。PUB は subject、オプションの reply subject、およびペイロードサイズを指定します。sid はその接続上のサブスクライバーIDであり、メッセージIDではありません。NATS Client Protocol

payload = '温度=23'.encode('utf-8')
assert len(payload) == 9
packet = b'PUB sensors.temp ' + str(len(payload)).encode('ascii') + b'\r\n' + payload + b'\r\n'
line, rest = packet.split(b'\r\n', 1)
size = int(line.split()[-1])
assert rest[:size].decode('utf-8') == '温度=23'
assert rest[size:] == b'\r\n'
print('NATS UTF-8 ペイロードのバイト区切りチェック通過')

NATS のフィルター * は1つのトークンを一致させ、末尾の > は1つ以上のトークンを一致させます。これを MQTT の # のゼロ層マッチングと直接交換しないでください。request/reply は reply subject を介してメッセージを返しますが、リクエストタイムアウトは、遠端で既に実行された後に発生する可能性があります。

JetStream：永続ストリームと状態を持つコンシューマー

JetStream stream は指定された subject のメッセージをキャプチャし、メモリ/ファイルストレージ、レプリケーション、保持ポリシーを設定できます。consumer は配信と確認の状態を保存します。pull などの消費方法をサポートしており、すべての NATS 機能をステートレス、プッシュのみ、または必ず最低レイテンシと分類することはできません。JetStream

パブリッシャーは JetStream publish acknowledgment を介してストリームへのエントリーを確認します。コンシューマーの ACK は consumer 状態を進めます。未確認のメッセージは設定に応じて再送され、AckWait と最大試行回数は処理時間と連携する必要があります。異なる retention policy は、確認後にメッセージを削除するかどうかの影響が異なり、「ACK は単なるオフセット」または「ACK は常に削除する」と一律に解釈することはできません。

メッセージデジュープリケーションウィンドウや二重確認などの機能により、プロトコルレベルの重複を減らすことができますが、対応する設定と API が必要です。ウィンドウ外の再送、アプリケーションによる新しいイベントの作成、または外部の副作用は、「exactly-once」という一言だけでは排除できません。JetStream 消費動作

バックログと回復要件による比較

入力が毎秒 1200 件、コンシューマーが安定して毎秒 1000 件を処理できると仮定すると、毎秒 200 件のバックログが発生し、10分後には 120,000 件のバックログになります。メッセージにメタデータを追加して平均 1 KiB とすると、約 117.2 MiB となり、レプリカコピーは含まれません。より軽量なワイヤープロトコルに変更しても、生産と消費の間の長期的な差は解消されません。

パフォーマンスレポートには、パケット長、バッチ、圧縮、確認、レプリケーション、ディスク、およびコンシューマーの作業負荷を同時に記載する必要があります。これらの条件がない「毎秒百万件、固定 0.1 ms」は選定根拠として機能せず、プロトコルの複雑さがそのままエンドツーエンドのレイテンシに直結するわけではありません。