DevOpsツールとベストプラクティスの習得 - DevOpsナレッジハブ DevOpsツールとベストプラクティスの習得 - DevOpsナレッジハブ

RabbitMQにおけるメッセージ損失防止：よくある落とし穴と解決策

確認応答、承認、永続キュー、DLQ、安全なリトライ動作を用いてRabbitMQのメッセージ損失を減らす実践的な方法。

RabbitMQにおけるメッセージ損失防止：よくある落とし穴と解決策

RabbitMQのメッセージ損失は、劇的なブローカーの障害によって引き起こされることはほとんどありません。多くの場合、公開または消費のパスに小さなギャップがあることが原因です。パブリッシャーはソケットへの書き込みがブローカーにメッセージが受け入れられたことを意味すると想定したり、コンシューマーはデータベースコミットが完了する前に確認応答を行ったり、キューは永続的であるにもかかわらず、それに送信されるメッセージが一時的なものであったりします。

RabbitMQの信頼性を確実に扱う最善の方法は、プロデューサーからブローカーへ、そしてブローカーからコンシューマーへとメッセージを追跡することです。各ステップで、「このメッセージはもう安全です」と言える権限を持つ主体を決定します。その決定はコード内で明示的に行い、監視で確認できるようにする必要があります。

メッセージライフサイクルと潜在的な損失ポイントの理解

解決策に飛び込む前に、RabbitMQの旅路においてメッセージがどこで失われる可能性があるかを理解することが不可欠です。

  • パブリッシャー側: パブリッシャーによって送信されたメッセージが、ネットワークの問題、ブローカーの利用不可、またはパブリッシャーのエラーにより、RabbitMQブローカーに到達しない可能性があります。
  • ブローカー側: メッセージがRabbitMQ内にある場合、ブローカーがメッセージをディスクに永続化する前にクラッシュしたり、メッセージが存在するキューが予期せず削除されたりすると、メッセージが失われる可能性があります。
  • コンシューマー側: コンシューマーがメッセージを受信しても、アプリケーションエラー、クラッシュ、または時期尚早な確認応答により処理に失敗し、メッセージがドロップされる可能性があります。

メッセージ損失防止のための主要なテクニック

RabbitMQは、メッセージの耐久性と信頼性を高めるためのいくつかの組み込み機能と推奨パターンを提供しています。これらを実装することは、データ損失を防ぐために重要です。

1. パブリッシャー確認応答

パブリッシャー確認応答は、メッセージが正常に受信され処理されたことをブローカーがパブリッシャーに通知するメカニズムを提供します。これは、パブリッシャーとブローカーの間でメッセージが消失しないようにするために重要です。

仕組み:

  1. パブリッシャーがRabbitMQにメッセージを送信します。
  2. RabbitMQはメッセージを受信すると、パブリッシャーに確認応答を送信するように設定できます。この確認応答は、メッセージが受け入れられたことを示します。
  3. RabbitMQがメッセージを受け入れられない場合（例：キューがいっぱい、ルーティングキーが無効）、否定応答（nack）を送信します。

設定:

パブリッシャー確認応答は、チャネルで confirm.select を設定することで有効になります。これにより、RabbitMQに対してチャネルが確認モードで動作する必要があることが通知されます。

例（Pythonの pika ライブラリを使用）:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange='',
        routing_key='my_queue',
        body='Hello, World!',
        properties=pika.BasicProperties(delivery_mode=2) # メッセージを永続化
    )
    print(" [x] Sent 'Hello, World!'")
    # 例外が発生しなければ、メッセージはブローカーによって確認された
except pika.exceptions.UnroutableMessageError as e:
    print(f"Message could not be routed: {e}")
except pika.exceptions.ChannelClosedByBroker as e:
    print(f"Channel closed by broker: {e}")
    # ここで接続またはブローカーの問題を処理
except Exception as e:
    print(f"An unexpected error occurred: {e}")

connection.close()

ベストプラクティス: パブリッシャー確認応答を使用する場合、nackやチャネルクローズを適切に処理するために、basic_publish 呼び出しの周りに常にエラーハンドリングを実装してください。

2. コンシューマー確認応答（Ack/Nack）

コンシューマー確認応答は、メッセージがコンシューマーに配信された後に失われないようにするために不可欠です。これにより、コンシューマーはメッセージが正常に処理されたかどうかをRabbitMQに通知できます。

確認応答の種類:

  • 自動確認応答（auto_ack=True）: RabbitMQはメッセージが配信されたとみなし、コンシューマーに送信するとすぐにキューから削除します。コンシューマーが処理前にクラッシュすると、メッセージは失われます。
  • 手動確認応答（auto_ack=False）: コンシューマーは、メッセージの処理が完了したときに明示的にRabbitMQに通知します。これにより、コンシューマーが失敗した場合に再配信が可能になります。

手動確認応答のフロー:

  1. コンシューマーがメッセージを受信します。
  2. コンシューマーがメッセージを処理します。
  3. 処理が成功した場合、コンシューマーはRabbitMQに basic_ack を送信します。
  4. 処理が失敗した場合、コンシューマーは以下のいずれかを実行できます。
    • requeue=True を指定して basic_nack（または basic_reject）を送信し、メッセージをキューに戻して別のコンシューマーが取得できるようにします。
    • requeue=False を指定して basic_nack（または basic_reject）を送信し、メッセージを破棄するか、Dead-Letter Exchange（DLX）に送信します。

例（Pythonの pika ライブラリを使用）:

import pika
import time

def callback(ch, method, properties, body):
    print(f" [x] Received {body}")
    try:
        # 処理をシミュレート
        if b'error' in body:
            raise Exception("Simulated processing error")
        # 処理が成功した場合:
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(" [x] Acknowledged message")
    except Exception as e:
        print(f"Processing failed: {e}")
        # メッセージを拒否して再キューイング
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        print(" [x] Rejected and requeued message")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='my_queue')

channel.basic_consume(queue='my_queue', on_message_callback=callback, auto_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

警告: メッセージが一貫して処理に失敗する場合、requeue=True を無制限に使用するとメッセージループが発生する可能性があります。ここでデッドレタリングが重要になります。

3. メッセージの永続化

デフォルトでは、RabbitMQのメッセージは一時的（transient）です。ブローカーが再起動すると、すべての一時メッセージは失われます。これを防ぐには、メッセージとキューを永続的（durable）として宣言する必要があります。

永続キュー:

キューを宣言するときに、durable パラメータを True に設定します。

channel.queue_declare(queue='my_durable_queue', durable=True)

永続メッセージ:

メッセージを公開するときに、delivery_mode プロパティを 2 に設定します。

channel.basic_publish(
    exchange='',
    routing_key='my_durable_queue',
    body='Persistent message',
    properties=pika.BasicProperties(delivery_mode=2) # 永続化
)

重要な注意点: メッセージの永続化は万能薬ではありません。メッセージは、キューに書き込まれた後にのみディスクに永続化されます。パブリッシャーがメッセージを送信したとみなす前に、メッセージがブローカーに到達し、永続キューに書き込まれたことを保証するには、パブリッシャー確認応答が依然として必要です。さらに、ディスク自体に障害が発生した場合、適切なディスク冗長性がなければ、永続化されたメッセージも失われる可能性があります。

4. デッドレタリング（DLX）

デッドレタリングは、正常に処理できないか、期限切れになったメッセージを処理するための強力なメカニズムです。これらのメッセージは破棄されたり無限に再キューイングされたりする代わりに、指定された「デッドレターエクスチェンジ」にルーティングされます。

デッドレタリングのシナリオ:

  • コンシューマーが requeue=False でメッセージを明示的に拒否する。
  • メッセージがTime-To-Live（TTL）設定により期限切れになる。
  • キューが最大長制限に達する。

設定:

  1. Dead-Letter Exchange（DLX）を宣言する: これはメッセージが送信される通常のエクスチェンジです。
  2. Dead-Letter Queue（DLQ）を宣言する: DLXにバインドされたキューです。
  3. 元のキューを設定する: デッドレターメッセージを生成する可能性のあるキューを宣言するときに、x-dead-letter-exchange および x-dead-letter-routing-key 引数を指定します。

例:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 1. DLXとDLQを宣言
channel.exchange_declare(exchange='my_dlx', exchange_type='topic')
channel.queue_declare(queue='my_dlq')
channel.queue_bind(queue='my_dlq', exchange='my_dlx', routing_key='dead')

# 2. DLX/DLQ引数を持つプライマリキューを宣言
channel.queue_declare(
    queue='my_processing_queue',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'my_dlx',
        'x-dead-letter-routing-key': 'dead'
    }
)

# 処理キューを目的のコンシューマーエクスチェンジにバインド（ある場合）
# 簡単のため、この例ではキューへの直接公開を想定

# コンシューマーで、メッセージが失敗した場合、拒否:
# channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

print("Queues and exchanges set up for dead-lettering.")
connection.close()

メッセージが my_processing_queue から requeue=False で拒否されると、ルーティングキー dead で my_dlx にルーティングされ、次に my_dlq にルーティングされます。その後、別のコンシューマーを設定して my_dlq を監視し、検査、再処理、またはアーカイブを行うことができます。

5. 高可用性とレプリケーション

重要なアプリケーションの場合、単一のRabbitMQノードは単一障害点となります。クラスタリングとレプリケートされたキュータイプは、ノード障害時のダウンタイムやデータ損失のリスクを軽減できますが、RabbitMQのバージョンとワークロードに応じて選択し、テストする必要があります。

  • クラスタリング: 複数のRabbitMQノードが単一のユニットとして連携します。キューはノード間で宣言できます。
  • レプリケートされたキュー: 最新のRabbitMQデプロイメントでは、レプリケートされた永続ワークロードにquorumキューが一般的に使用されます。古いクラシックなHAパターンは、新しいユースケースの前に現在のRabbitMQガイダンスと照らし合わせて評価する必要があります。

レプリケーションは可用性を向上させますが、ネットワークとディスクの作業も追加します。重要なワークフローで信頼する前に、パブリッシャー確認応答のレイテンシ、フェイルオーバー動作、コンシューマーの再配信をテストしてください。

実際に必要な信頼性契約

RabbitMQでのメッセージ損失を防ぐことは、各キューの契約を書き留めると推論しやすくなります。すべてのキューが同じ保護に値するわけではありません。キャッシュ無効化イベントを運ぶキューは、キャッシュの期限切れや再構築が可能なため、メッセージの欠落を許容する場合があります。支払いキャプチャリクエスト、パスワードリセットメールリクエスト、出荷ステータス変更、監査イベントを運ぶキューは、通常、はるかに強力な契約を必要とします。

契約は4つの明白な質問に答える必要があります。

  • パブリッシャーが送信後にクラッシュした場合、安全に再試行できますか？
  • RabbitMQが再起動した場合、メッセージはまだ存在している必要がありますか？
  • コンシューマーが作業の途中でクラッシュした場合、メッセージは再試行されるべきですか？
  • メッセージが失敗し続ける場合、それはどこに行き、誰がそれを確認しますか？

実際のメッセージ損失インシデントのほとんどは、これらの質問のいずれにも答えられなかったために発生します。コードはキューを使用しているかもしれませんが、システムには「送信済み」または「処理済み」の意味についての合意がありません。

より安全なパブリッシャーは、ブローカーが確認した後にのみメッセージを送信済みとして扱います。より安全なキューは、メッセージがブローカーの再起動後も存続する必要がある場合に永続的です。より安全なメッセージは、内容が重要な場合に永続的として公開されます。より安全なコンシューマーは、永続的な副作用が完了した後にのみ確認応答を行います。より安全な障害パスは、無限にループする代わりに、ポイズンメッセージをデッドレターキューに送信します。

これは多く聞こえるかもしれませんが、実際には、すべての重要なワークフローに適用できる短いチェックリストになります。

実際の障害パターン：早期確認応答

私が見る最も一般的なRabbitMQメッセージ損失バグは、特殊なものではありません。次のようになります。

  1. コンシューマーが注文イベントを受信します。
  2. コンシューマーはすぐにメッセージを確認応答します。
  3. コンシューマーは外部の請求APIを呼び出します。
  4. プロセスがクラッシュするか、APIリクエストがタイムアウトします。

RabbitMQは指示されたことを正確に実行しました。コンシューマーは「完了しました」と言ったため、ブローカーはメッセージを削除しました。ビジネスオペレーションは完了していませんでしたが、ブローカーはそれを知る方法がありませんでした。

修正方法は、確認応答を元に戻せない作業の後に移動することです。

def callback(ch, method, properties, body):
    try:
        event = parse_order_event(body)
        charge_id = charge_customer(event)
        save_charge_result(event["order_id"], charge_id)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except TemporaryBillingError:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    except InvalidOrderError:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

それでもまだ1つの微妙な問題が残ります。コンシューマーが課金結果を保存した後、basic_ack の前にクラッシュした場合はどうなりますか？RabbitMQはメッセージを再配信します。これは損失ではありませんが、重複処理になる可能性があります。信頼性の高いRabbitMQコンシューマーは、通常、べき等である必要があります。メッセージID、注文ID、またはビジネスキーを使用して、同じメッセージを繰り返しても実際の副作用が繰り返されないようにします。

たとえば、一意制約のあるテーブルに order_id と charge_id を書き込むコンシューマーは、再配信を安全に処理できます。2回目の実行では、レコードがすでに存在することを確認し、再度課金することなくメッセージを確認応答します。

重要なメッセージにはパブリッシャー確認応答は必須

パブリッシャー確認応答がない場合、パブリッシャーはソケットにバイトを書き込んだことだけを知っています。RabbitMQがメッセージを受け入れたか、ルーティングしたか、永続化したか、ブローカーが処理する前に接続が失われたかどうかはわかりません。

ファイアアンドフォーゲットのテレメトリでは、それが許容される場合があります。ビジネスアクションを表すワークキューでは、それだけでは不十分です。

適切なパブリッシャーパスは通常、次の3つのことを行います。

  • チャネルでパブリッシャー確認応答を有効にします。
  • 重要なメッセージを永続的としてマークします。
  • mandatory=True または代替エクスチェンジを使用して、ルーティング不可能なメッセージを処理します。

ルーティング不可能なメッセージの部分は見落とされがちです。どのキューにも一致しないルーティングキーでエクスチェンジに公開すると、RabbitMQは公開を受け入れることができますが、通知を要求しない限りどこにもルーティングしません。アプリケーションの観点からは、メッセージ損失のように見えます。

pika では、正確な動作はチャネルモードと例外処理に依存しますが、意図は次のとおりです。

channel.confirm_delivery()

channel.basic_publish(
    exchange="orders",
    routing_key="created",
    body=payload,
    mandatory=True,
    properties=pika.BasicProperties(
        delivery_mode=2,
        message_id=order_id,
        content_type="application/json",
    ),
)

公開が失敗した場合は、注意して再試行してください。再試行ループは、盲目的に重複したビジネスイベントを作成するべきではありません。最初にアプリケーションデータベースに送信イベントを保存し、公開してから、確認後に公開済みとしてマークします。この「アウトボックス」パターンは、データベースコミットとメッセージ公開の間の厄介なギャップを処理するため、一般的です。

永続化には3つの要素がある

RabbitMQの耐久性は、複数のスイッチがあるため、しばしば誤解されます。

エクスチェンジは、再起動後も存在することが期待される場合、永続的である必要があります。キューは、再起動後も存在することが期待される場合、永続的である必要があります。メッセージは、その内容が再起動後も存続することが期待される場合、永続的である必要があります。

これらのいずれかを省略すると、驚くことがあります。非永続キューに送信された永続メッセージは、キューを永続的にしません。一時メッセージを受信する永続キューは、再起動中にそれらの一時メッセージを失う可能性があります。永続エクスチェンジと永続キューは、デプロイメントがトポロジを誤って削除および再作成する場合には役に立ちません。

スタートアップコードまたはインフラストラクチャ自動化を使用して、トポロジを一貫して宣言します。

channel.exchange_declare(
    exchange="orders",
    exchange_type="topic",
    durable=True,
)

channel.queue_declare(
    queue="order_processing",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "orders.dlx",
        "x-dead-letter-routing-key": "order_processing.failed",
    },
)

channel.queue_bind(
    queue="order_processing",
    exchange="orders",
    routing_key="created",
)

永続化はブローカー再起動時の損失を減らしますが、バックアップ、ディスク冗長性、quorumレプリケーション、パブリッシャー確認応答に代わるものではありません。また、コストも伴います。永続メッセージはディスク作業を必要とし、高い公開レートは低速ストレージをすぐに露呈する可能性があります。これは、重要なデータに対して永続化を避ける理由にはなりません。ラップトップのベンチマークが本番環境に適用されると想定するのではなく、実際のワークロードをテストする理由です。

ポイズンメッセージループを作成しないリトライ

basic_nack(..., requeue=True) は一時的な障害には便利ですが、危険になる可能性があります。メッセージが常に失敗する場合、何度も配信されます。ブローカーは再配信に作業を費やします。コンシューマーは失敗に作業を費やします。後ろにある正常なメッセージは、本来よりも長く待たされる可能性があります。

より良いパターンは、クイックリトライと遅延リトライ、そして最終的な失敗を分離することです。

1つの簡単なセットアップ：

  • 最初の失敗：エラーが明らかに一時的なものである場合は、1回だけ再キューイングします。
  • 繰り返しの失敗：requeue=False で拒否します。
  • デッドレターキュー：ヘッダーとルーティングコンテキストとともに失敗したメッセージを保存します。
  • 再生ツール：オペレーターまたはスケジュールされたジョブが、根本原因が修正された後に検査して再公開できるようにします。

遅延リトライの場合、多くのチームはTTLと元のキューに戻すデッドレターエクスチェンジを備えたリトライキューを使用します。これにより、失敗している依存関係に毎ミリ秒ヒットすることなく、回復する時間を与えることができます。

ヘッダーには注意してください。RabbitMQは x-death などのデッドレターメタデータを追加します。コンシューマーはそれを読み取って、メッセージがすでにリトライ回数が多すぎるかどうかを判断できます。コンシューマープロセス内のメモリのみに依存しないでください。その状態は再起動時に消えます。

キューを信頼する前の運用チェック

コードを設定した後、意図的に厄介なケースをテストします。

メッセージを公開中にコンシューマーを停止します。キューの深さが上昇し、永続的であることを意図している場合、メッセージはブローカー再起動後も残る必要があります。コンシューマーを再起動し、キューをドレインすることを確認します。

処理中にコンシューマーを強制終了します。手動確認応答を使用すると、チャネルが閉じられた後、処理中のメッセージは再び準備完了状態になる必要があります。メッセージが消えた場合は、確認応答が早すぎるか、どこかで自動確認応答を使用しています。

不正なルーティングキーで公開します。パブリッシャーは、リターン、確認関連のエラー、または代替エクスチェンジパスを通じて失敗を認識する必要があります。公開呼び出しが成功したように見え、メッセージがどこにも届かない場合、ルーティングのセーフティネットは不完全です。

既知の不良メッセージでデッドレターキューを満たします。なぜ失敗したか、何回試行されたか、安全に再生できるかどうかを確認できる必要があります。所有者のいないDLQは、メッセージを失うためのより遅い方法にすぎません。

テスト中にこれらのメトリクスを監視します。

  • messages_ready: コンシューマーを待っているメッセージ。
  • messages_unacknowledged: 配信されたがまだ確認応答されていないメッセージ。
  • クライアント側からの公開確認応答レイテンシ。
  • コンシューマーエラー率とリトライ回数。
  • デッドレターキューの深さ。
  • メモリとディスクのアラーム。

目標は、RabbitMQがすべてのビジネス成果を魔法のように保証することではありません。目標は、すべての障害を可視化し、回復可能にすることです。

最終的な信頼性チェック

すべての重要なRabbitMQワークフローについて、パブリッシャーがブローカーの確認を待つこと、エクスチェンジとキューが再起動後も存続する必要がある場合に永続的であること、メッセージ自体がその内容が重要な場合に永続的であること、コンシューマーが実際の作業が完了した後にのみ確認応答を行うことを確認します。次に、障害ケース（不正なルーティングキー、ブローカー再起動、コンシューマークラッシュ、繰り返しの処理失敗、DLQ再生）をテストします。

これらのテストがビジネスの期待どおりに動作する場合、RabbitMQがメッセージを安全に保つことを単に期待しているだけではありません。何かが壊れたときの回復パスがあります。