PHP 메시지 대기열의 메시지 확인 및 재시도 메커니즘 소개-PHP 튜토리얼-php.cn

PHP 메시지 대기열의 메시지 확인 및 재시도 메커니즘 소개

PHPz

풀어 주다： 2023-07-09 20:58:01

원래의

1808명이 탐색했습니다.

PHP 메시지 대기열의 메시지 확인 및 재시도 메커니즘 소개

인터넷 애플리케이션의 개발로 인해 높은 동시성 및 대규모 트래픽 시나리오에 직면한 전통적인 직접 요청 방법은 더 이상 요구 사항을 충족할 수 없습니다. 분리 및 비동기 처리 기술 솔루션으로서 메시지 큐는 많은 기업 수준 애플리케이션에서 널리 사용됩니다. PHP 메시지 큐에서 메시지 확인과 재시도 메커니즘은 두 가지 매우 중요한 개념입니다. 이 기사에서는 이를 자세히 소개하고 해당 코드 예제를 제공합니다.

메시지 확인 메커니즘

메시지 대기열에서 메시지 확인 메커니즘은 소비자가 메시지를 받은 후 확인 메시지를 메시지 대기열에 보내 메시지가 성공적으로 처리되었음을 메시지 대기열에 알리는 것을 의미합니다. 소비자가 메시지를 처리할 때 예외가 발생하거나 처리가 실패하면 메시지 대기열은 확인 정보를 받지 못하고 메시지 처리가 실패한 것으로 간주하여 메시지를 다른 소비자에게 재배포하거나 프로세스를 다시 시도합니다.

메시지 확인 메커니즘의 구현은 다음 두 가지 측면을 고려해야 합니다.
(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'])를 호출합니다. 메시지 대기열에 확인 정보를 보냅니다. $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag'])来发送确认信息给消息队列。

对于没有收到确认信息的消息，消息队列根据具体的策略来处理。一种常见的处理方式是设置消息的过期时间，如果消息在一定时间内没有收到确认信息，则认为该消息处理失败，消息队列将重新分发该消息给其他消费者。

消息重试机制

消息重试机制是指在消息处理失败后，对该消息进行重试处理的机制。在消息队列中，消息的重试可以基于以下两种方式：
(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();

로그인 후 복사

在上面的代码中，声明了一个名称为"task_queue"的队列，使用$channel->basic_qos(null, 1, null);设置每次只分发一条消息，并在回调函数内模拟了消息处理失败的情况。当处理失败时，调用示例代码中的$msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true);

확인 정보를 받지 못한 메시지의 경우 메시지 대기열에서 특정 전략에 따라 메시지를 처리합니다. 일반적인 처리 방법은 메시지의 만료 시간을 설정하는 것입니다. 메시지가 일정 시간 내에 확인 정보를 받지 못하면 메시지 처리가 실패한 것으로 간주되며 메시지 큐는 메시지를 다른 소비자에게 재분배합니다.

$channel->basic_qos(null, 1, null);한 번에 하나의 메시지만 배포하도록 설정하고 콜백 함수에서 메시지 처리 실패를 시뮬레이션합니다. 처리 실패 시 샘플 코드에서 $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true);를 호출하여 Retry를 보냅니다. 메시지 대기열에 메시지를 보냅니다. 🎜🎜메시지 확인 메커니즘과 재시도 메커니즘을 통해 PHP 메시지 큐는 메시지의 신뢰성과 처리 효율성을 보장할 수 있습니다. 개발자는 더 나은 사용자 경험과 시스템 성능을 제공하기 위해 실제 요구 사항에 따라 적절한 메시지 확인 및 재시도 전략을 선택할 수 있습니다. 🎜

위 내용은 PHP 메시지 대기열의 메시지 확인 및 재시도 메커니즘 소개의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!