翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Amazon MQ for RabbitMQ でのメッセージの耐久性と信頼性に関するベストプラクティス

アプリケーションを本番環境に移行する前に、メッセージの損失とリソースの過剰使用を防ぐための以下のベストプラクティスを完了してください。

ステップ 1: 永続メッセージと耐久性のあるキューを使用する

永続メッセージは、ブローカーがクラッシュまたは再起動するという状況におけるデータ損失の防止に役立ちます。永続メッセージは、到着するとすぐにディスクに書き込まれますが、レイジーキューとは異なり、ブローカーがより多くのメモリを必要とする場合を除き、永続メッセージはメモリとディスクの両方にキャッシュされます。より多くのメモリが必要な場合は、ディスクへのメッセージの保存を管理する RabbitMQ ブローカーメカニズム (一般に永続レイヤーと呼ばれます) によって、メモリからメッセージが削除されます。

メッセージの永続性を有効にするには、キューを durable として宣言し、メッセージ配信モードを persistent に設定できます。以下の例は、RabbitMQ Java クライアントライブラリを使用した持続キューの宣言を示しています。AMQP 0-9-1 を使用している場合は、配信モード「2」を設定することで、メッセージを永続としてマークできます。

boolean durable = true;
channel.queueDeclare("my_queue", durable, false, false, null);

キューを持続キューとして設定したら、以下の例にあるように、MessageProperties を PERSISTENT_TEXT_PLAIN に設定することによって永続メッセージをキューに送信できます。

import com.rabbitmq.client.MessageProperties;

channel.basicPublish("", "my_queue",
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());

ステップ 2: パブリッシャーの確認とコンシューマー配信の確認を設定する

ブローカーにメッセージが送信されたことを確認するプロセスは、パブリッシャーの確認と呼ばれます。パブリッシャーは、メッセージが確実に格納されたときにアプリケーションに通知します。パブリッシャーの確認は、ブローカーに格納されるメッセージの割合を制御するためにも役立ちます。パブリッシャーが確認しないと、メッセージが正常に処理されたという確認は行われず、ブローカーは処理できないメッセージを削除する可能性があります。

同様に、クライアントアプリケーションはメッセージの配信と消費の確認をブローカーに返送します。これはコンシューマーの配信承認と呼ばれます。RabbitMQ ブローカーの使用時にデータの安全性を確保するには、確認と承認の両方が不可欠です。

コンシューマーの配信承認は、通常クライアントアプリケーションで設定されています。AMQP 0-9-1 を使用している場合は、basic.consume メソッドを設定することで承認を有効化できます。AMQP 0-9-1 クライアントでは、confirm.select メソッドを送信してパブリッシャーの確認を設定することもできます。

通常、配信承認はチャネルで有効化されます。例えば、RabbitMQ Java クライアントライブラリの使用時には、以下の例にあるように、Channel#basicAck を使用してシンプルな basic.ack 肯定承認をセットアップできます。

// this example assumes an existing channel instance

boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "a-consumer-tag",
     new DefaultConsumer(channel) {
         @Override
         public void handleDelivery(String consumerTag,
                                    Envelope envelope,
                                    AMQP.BasicProperties properties,
                                    byte[] body)
             throws IOException
         {
             long deliveryTag = envelope.getDeliveryTag();
             // positively acknowledge a single delivery, the message will
             // be discarded
             channel.basicAck(deliveryTag, false);
         }
     });

consumer_timeout を設定すると、コンシューマーから配信承認が届かない状況を検出できます。コンシューマーがタイムアウト値の時間内に承認を送信しない場合、チャネルは閉じられ、PRECONDITION_FAILED が発生します。エラーを診断するには、UpdateConfiguration API を使用して consumer_timeout 値を大きくします。

ステップ 3: キューを短くする

クラスターデプロイでは、多数のメッセージを持つキューがリソースの過剰な使用につながる場合があります。ブローカーが過剰に使用されているときは、Amazon MQ for RabbitMQ ブローカーの再起動がパフォーマンスをさらに低下させる原因となる可能性があります。過剰に使用されているブローカーが再起動されると、REBOOT_IN_PROGRESS 状態のまま応答しなくなることがあります。

Amazon MQ はメンテナンスウィンドウ中、すべてのメンテナンス作業を一度に 1 ノードずつ実行して、ブローカーが動作可能な状態を維持することを確実にします。その結果、各ノードが操作を再開するときに、キューが同期する必要が生じる場合があります。同期中、ミラーにレプリケートする必要があるメッセージは、バッチで処理されるように、対応する Amazon Elastic Block Store (Amazon EBS) ボリュームからメモリにロードされます。メッセージをバッチで処理することにより、キューの同期が速くなります。

キューを短くし、メッセージを小さくしておくと、キューが正常に同期し、期待通りに操作を再開します。ただし、バッチ内のデータ量がノードのメモリ制限に近づいた場合は、ノードが高メモリアラームを発し、キューの同期を一時停止します。メモリ使用量は、CloudWatch で RabbitMemUsed および RabbitMqMemLimit のブローカーノードメトリクスを比較することで確認できます。同期は、メッセージが消費もしくは削除される、またはバッチ内のメッセージの数が減るまで完了できません。

クラスターデプロイのためにキューの同期化が一時停止される場合は、メッセージを消費または削除して、キュー内のメッセージの数を減らすことをお勧めします。キュー深度が減少し、キューの同期が完了すると、ブローカーのステータスが RUNNING に変更されます。一時停止されたキューの同期を解決するには、キューの同期のバッチサイズを小さくするポリシーを適用することも可能です。

また、自動削除ポリシーと TTL ポリシーを定義すると、リソースの使用量をプロアクティブに削減するとともに、コンシューマーからの NACK を最小限に抑えることができます。ブローカーへのメッセージの再キュー処理は CPU 負荷が高いため、大量の NACK が発生するとブローカーのパフォーマンスに影響する可能性があります。