PHP メッセージキューのメッセージ確認と再試行メカニズムの概要

PHP メッセージ キューのメッセージ確認および再試行メカニズムの概要

インターネット アプリケーションの開発に伴い、高い同時実行性と大規模なトラフィックのシナリオに直面すると、従来の直接リクエスト方法ではニーズを満たすことができなくなりました。メッセージ キューは、分離および非同期処理の技術ソリューションとして、多くのエンタープライズ レベルのアプリケーションで広く使用されています。 PHP メッセージ キューでは、メッセージ確認と再試行メカニズムの 2 つの非常に重要な概念がこの記事で詳しく紹介され、対応するコード例が示されます。

1. メッセージ確認メカニズム

メッセージ キューにおけるメッセージ確認メカニズムとは、コンシューマがメッセージを受信した後、メッセージ キューに確認メッセージを送信して通知することを意味します。メッセージが正常に処理されたメッセージ キュー。コンシューマがメッセージを処理するときに例外が発生したり、処理が失敗した場合、メッセージ キューは確認情報を受信せず、メッセージの処理が失敗したものとみなし、メッセージを他のコンシューマに再配布するか、プロセスを再試行します。

メッセージ確認メカニズムの実装では、次の 2 つの側面を考慮する必要があります。
(1) メッセージ コンシューマーがメッセージ キューに確認情報を送信する方法
(2) メッセージ キューがメッセージを処理する方法確認情報の受信に失敗しました。 メッセージ

PHP メッセージ キューでは、通常、メッセージ確認メカニズムの実装に AMQP (Advanced Message Queuing Protocol) プロトコルが使用されます。以下は、rabbitmq の使用例です。

<?php
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('hello', false, false, false, false);

echo " [*] Waiting for messages. To exit press CTRL+C
";

$callback = function ($msg) {
    echo ' [x] Received ', $msg->body, "
";
    // 处理消息的业务逻辑

    // 发送确认信息给消息队列
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};

$channel->basic_consume('hello', '', false, false, false, false, $callback);

while (count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connection->close();

上記のコードでは、メッセージ キュー接続とチャネルが作成され、「hello」という名前のキューが宣言されています。コールバック関数では、受信したメッセージに対して業務処理を行った後、$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag'])を呼び出して、確認情報をメッセージキューに送信します。

確認情報を受信して​​いないメッセージの場合、メッセージ キューは特定の戦略に従ってメッセージを処理します。一般的な処理方法は、メッセージの有効期限を設定することです。メッセージが一定期間内に確認情報を受信しない場合、メッセージの処理は失敗したとみなされ、メッセージ キューはメッセージを他のコンシューマに再配布します。

2. メッセージ再試行メカニズム

メッセージ再試行メカニズムとは、メッセージ処理が失敗した後にメッセージを再試行するためのメカニズムを指します。メッセージ キューでは、メッセージの再試行は次の 2 つの方法に基づいて行うことができます。
(1) 固定再試行回数: 処理に失敗したメッセージの場合、メッセージ キューはメッセージをコンシューマに再配布し、毎回再試行します。再試行時にカウンタを増やす。カウンタが固定の再試行回数に達すると、メッセージ キューは再試行を行わなくなり、メッセージを特定の失敗キューに送信して手動介入を待ちます。
(2) 指数ベースの再試行時間: 処理に失敗したメッセージの場合、メッセージ キューはメッセージをコンシューマに再配布し、指数ベースの方法に基づいて各再試行の時間間隔を決定します。通常、短期間に多数の再試行が行われることを避け、システムの負荷を軽減するために、各再試行間の時間間隔は指数関数的に増加します。

以下は、rabbitmq のメッセージ再試行メカニズムの使用例です。

<?php
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

echo " [*] Waiting for messages. To exit press CTRL+C
";

$callback = function ($msg) {
    echo ' [x] Received ', $msg->body, "
";
    // 模拟消息处理失败的情况
    if (rand(0, 10) < 3) {
        // 发送重试信息给消息队列
        $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true);
    } else {
        // 处理消息的业务逻辑

        // 发送确认信息给消息队列
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    }
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);

while (count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connection->close();

上記のコードでは、$channel->basic_qos( を使用して、「task_queue」という名前のキューが宣言されています。 null, 1, null);一度に 1 つのメッセージのみを配布し、コールバック関数でメッセージ処理の失敗をシミュレートするように設定します。処理が失敗した場合は、サンプルコードの $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true); を呼び出してリトライを送信します。メッセージ キューにメッセージを送信してみます。

PHP メッセージ キューは、メッセージ確認メカニズムと再試行メカニズムを通じて、メッセージの信頼性と処理の効率を確保できます。開発者は、実際のニーズに基づいて適切なメッセージ確認および再試行戦略を選択し、より良いユーザー エクスペリエンスとシステム パフォーマンスを提供できます。

以上がPHP メッセージキューのメッセージ確認と再試行メカニズムの概要の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。