Verbrauchen nicht anerkennen Nachrichten von RabbitMq

Wenn die Meldung wurde nicht anerkannt und die Anwendung fehlschlägt, wird es nachgeliefert und automatisch redelivered - Eigenschaft auf den Umschlag gesetzt werden true (es sei denn, Sie konsumieren mit no-ack = true flag).

UPD:

Müssen Sie nack Nachricht mit durchgangsschein Flagge in deinem catch-block

    try {
        //Do some business logic
    } catch (Exception $ex) {
        //Log exception
        return $queue->nack($msg->getDeliveryTag(), AMQP_REQUEUE);
    }

Vorsicht unendlich nacked Nachrichtenwährend eine Nachlieferung zählen nicht umgesetzt RabbitMQ und in der AMQP-Protokoll überhaupt.

Wenn Sie nicht zu verwirren wollen mit solchen Nachrichten und wollen einfach hinzufügen einiger Verspätung möchten Sie vielleicht fügen Sie einige sleep() oder usleep() vor nack Methode aufrufen, aber es ist nicht eine gute Idee überhaupt.

Gibt es mehrere Techniken zum Umgang mit Zyklus wieder problem:

1. Verlassen Sie sich auf Dead Letter-Austausch

• Vorteile: zuverlässig, standard, klar
• Nachteile: erfordern zusätzliche Logik

2. Verwenden pro Nachricht oder pro Warteschlange TTL

• Vorteile: einfach zu implementieren, auch standard, klar
• Nachteile: lange Warteschlangen, Sie können Locker einige Nachricht

Beispiele (beachten Sie, dass für die queue-ttl passieren wir nur die Nummer und Nachricht ttl - alles, was wird die numerische Zeichenfolge):

2.1 Pro Nachricht ttl:

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish(
    'message at ' . microtime(true),
    null,
    AMQP_NOPARAM,
    array(
        'expiration' => '1000'
    )
);

2.2. Pro Warteschlange ttl:

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->setArgument('x-message-ttl', 1000);
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish('message at ' . microtime(true));

3. Halten redelivers zählen oder Links redelivers Anzahl (aka hop-limit oder ttl im IP-stack) im message-body oder Header

• Vorteile: geben Sie zusätzliche Kontrolle über Nachrichten Lebenszeit auf application level
• Nachteile: erheblicher Aufwand, während Sie auf Nachricht ändern und es wieder zu veröffentlichen, die für die Anwendung spezifischen, nicht klar

Code:

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish(
    'message at ' . microtime(true),
    null,
    AMQP_NOPARAM,
    array(
        'headers' => array(
            'ttl' => 100
        )
    )
);

$queue->consume(
    function (AMQPEnvelope $msg, AMQPQueue $queue) use ($exchange) {
        $headers = $msg->getHeaders();
        echo $msg->isRedelivery() ? 'redelivered' : 'origin', ' ';
        echo $msg->getDeliveryTag(), ' ';
        echo isset($headers['ttl']) ? $headers['ttl'] : 'no ttl' , ' ';
        echo $msg->getBody(), PHP_EOL;

        try {
            //Do some business logic
            throw new Exception('business logic failed');
        } catch (Exception $ex) {
            //Log exception
            if (isset($headers['ttl'])) {
                //with ttl logic

                if ($headers['ttl'] > 0) {
                    $headers['ttl']--;

                    $exchange->publish($msg->getBody(), $msg->getRoutingKey(), AMQP_NOPARAM, array('headers' => $headers));
                }

                return $queue->ack($msg->getDeliveryTag());
            } else {
                //without ttl logic
                return $queue->nack($msg->getDeliveryTag(), AMQP_REQUEUE); //or drop it without requeue
            }

        }

        return $queue->ack($msg->getDeliveryTag());
    }
);

Kann es einige andere Wege, um eine bessere Kontrolle Nachricht redelivers fließen.

Abschluss: es gibt keine Wunderwaffe ist. Sie haben zu entscheiden, welche Lösung Ihren Bedarf das beste oder etwas anderen, aber vergesst nicht es zu teilen hier 😉

InformationsquelleAutor der Antwort pinepain