Introduction au mécanisme de confirmation et de nouvelle tentative de message dans la file d'attente de messages PHP

Introduction au mécanisme de confirmation et de nouvelle tentative de message dans la file d'attente de messages PHP

Avec le développement des applications Internet, face à des scénarios de forte concurrence et de trafic important, la méthode traditionnelle de requête directe ne peut plus répondre aux besoins. En tant que solution technique de découplage et de traitement asynchrone, la file d'attente de messages est largement utilisée dans de nombreuses applications au niveau de l'entreprise. Dans la file d'attente de messages PHP, la confirmation des messages et le mécanisme de nouvelle tentative sont deux concepts très importants. Cet article les présentera en détail et donnera des exemples de code correspondants.

1. Mécanisme de confirmation des messages

Dans la file d'attente des messages, le mécanisme de confirmation des messages signifie qu'une fois que le consommateur a reçu le message, il envoie un message de confirmation à la file d'attente des messages pour informer la file d'attente des messages que le message a été traité avec succès. Si une exception se produit ou si le traitement échoue lorsqu'un consommateur traite un message, la file d'attente des messages ne recevra pas d'informations de confirmation et considérera que le traitement du message a échoué et redistribuera le message à d'autres consommateurs ou réessayera le processus.

La mise en œuvre du mécanisme de confirmation des messages doit prendre en compte les deux aspects suivants :
(1) Comment le consommateur du message envoie les informations de confirmation à la file d'attente des messages
(2) Comment la file d'attente des messages gère les messages qui n'ont pas reçu d'informations de confirmation

Dans la file d'attente de messages PHP, le protocole AMQP (Advanced Message Queuing Protocol) est généralement utilisé pour implémenter le mécanisme de confirmation des messages. Voici un exemple d'utilisation de 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();

Dans le code ci-dessus, une connexion et un canal de file d'attente de messages sont créés, et une file d'attente nommée "hello" est déclarée. Dans la fonction de rappel, après avoir effectué le traitement métier sur le message reçu, appelez $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']) pour envoyer des informations de confirmation à la file d'attente des messages. $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag'])来发送确认信息给消息队列。

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

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();

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

Pour les messages qui n'ont pas reçu d'informations de confirmation, la file d'attente des messages les traitera selon des stratégies spécifiques. Une méthode de traitement courante consiste à définir le délai d'expiration du message. Si le message ne reçoit pas d'informations de confirmation dans un certain délai, le traitement du message est considéré comme ayant échoué et la file d'attente des messages redistribuera le message aux autres consommateurs.

Mécanisme de nouvelle tentative de message🎜🎜🎜Le mécanisme de nouvelle tentative de message fait référence à un mécanisme permettant de réessayer le message après l'échec du traitement du message. Dans la file d'attente des messages, les nouvelles tentatives de messages peuvent être basées sur les deux méthodes suivantes : 🎜(1) Nombre fixe de tentatives : pour les messages qui ne parviennent pas à être traités, la file d'attente des messages redistribuera le message au consommateur et réessayera à chaque fois lorsque le compteur atteint un nombre fixe de tentatives, la file d'attente des messages ne réessayera plus, mais enverra le message à une file d'attente d'échec spécifique pour attendre une intervention manuelle. 🎜(2) Temps de nouvelle tentative exponentiel : pour les messages qui ne parviennent pas à être traités, la file d'attente des messages redistribuera le message au consommateur et déterminera l'intervalle de temps pour chaque nouvelle tentative en fonction de la méthode exponentielle. Habituellement, l'intervalle de temps entre chaque nouvelle tentative est augmenté de façon exponentielle pour éviter un grand nombre de tentatives sur une courte période et réduire la charge du système. 🎜🎜Ce qui suit est un exemple d'utilisation du mécanisme de nouvelle tentative de message de Rabbitmq : 🎜rrreee🎜Dans le code ci-dessus, une file d'attente nommée "task_queue" est déclarée, en utilisant $channel->basic_qos(null, 1, null);Configuré pour distribuer un seul message à la fois et simuler l'échec du traitement du message dans la fonction de rappel. Lorsque le traitement échoue, appelez $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true); dans l'exemple de code pour envoyer Réessayez messages dans la file d'attente des messages. 🎜🎜Grâce au mécanisme de confirmation des messages et au mécanisme de nouvelle tentative, la file d'attente des messages PHP peut garantir la fiabilité des messages et l'efficacité du traitement. Les développeurs peuvent choisir des stratégies appropriées de confirmation de message et de nouvelle tentative en fonction des besoins réels pour offrir une meilleure expérience utilisateur et de meilleures performances du système. 🎜

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!