phprabbitmqphp-amqplib

How in RabbitMQ and PHP return task back to the queue?


How can I return message back to the queue if processing result did not suit me. Found only information about message acknowledgments but I think that it does not suit me. I need that if as a result of processing I get the parameter RETRY message is added back to the queue. And then this worker or another one picks it up again and tries to process it.

For example:

<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;

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

$connection = new AMQPStreamConnection($AMQP);
$channel = $connection->channel();

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

$callback = function($msg) {
    $condition = json_decode($msg->body);

    if (!$condition) {
        # return to the queue
    }
};

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

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

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

Solution

  • set auto no_ack flag to false

    queue: Queue from where to get the messages
    consumer_tag: Consumer identifier
    no_local: Don't receive messages published by this consumer.
    no_ack: Tells the server if the consumer will acknowledge the messages.
    exclusive: Request exclusive consumer access, meaning only this consumer can access the queue
    nowait:
    callback: A PHP Callback

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

    you must use acknowledgments , if your proccess not work you can ignore ack

    <?php
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    
    echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
    
    $connection = new AMQPStreamConnection($AMQP);
    $channel = $connection->channel();
    
    $channel->queue_declare('test', false, false, false, false);
    
    $callback = function($message) {
        $condition = json_decode($message->body);
         
        if (!$condition) {
            // return to the queue 
            $message->delivery_info['channel']->basic_nack($message->delivery_info['delivery_tag']);
        }else{
            // send ack , remove from queue
            $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        }
    };
    
    $channel->basic_consume('test', '', false, false, false, false, $callback);
    
    while(count($channel->callbacks)) {
        $channel->wait();
    }
    
    $channel->close();
    $connection->close();
    

    Of course with this approach you will face with the message always in the head of the queue, there is also another possibility, if you really want to have a track of retry you can follow the below approach

    defining a queue for retry, preferably your queue-name -retry and define a dead-letter queue preferably: -dlq

    Then you can do something like below: How to set up -retry queue: this is the most important part of it. you need to declare queue with the following features:

    x-dead-letter-exchange: should be same as your main queue routing key
    x-dead-letter-routing-key: should be same as your main queue routing key
    x-message-ttl: the delay between retries

    the codes are sudo code, please do not copy-paste, this is just a hint to give you the idea about it

    $maximumRetry = 5;
    $callback = function($message) {
        $body = json_decode($message->body);
        try { 
            // process result is your condition
            $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        } catch(Exception $e) {
            // return to the queue 
            $body['try_attempt'] = !empty($body['try_attempt'])? int($body['try_attempt']) + 1: 1 
            if ($body['try_attempt'] >= $maximumRetry ){
                $message->delivery_info['channel']->basic_nack($message->delivery_info['delivery_tag']);
                return
            }
            $msg = new AMQPMessage(json_encode($message));
    
            $channel->basic_publish($msg, '', 'test-retry');
        }
    };
    

    We gonna need 3 queues for retying.

    ------------- Update -------------

    Quorum queues are providing ability out of the box so in the consumer, you can understand how many times each message was retried and you can also define a dead-letter queue for it easily, for more information you can read more about quorom queues and poison message handling