Reconnection guidelines - graceful dropped connection handling on consumer #444

aaronpeterson · 2016-11-01T16:11:35Z

Hoping to clarify best practices in reconnecting a consumer after a dropped connection, service outage, etc. Following this basic pattern:

class MyConsumer {

    protected $amqpChannel;
    protected $amqpConnection;

    public function connect()
    {
        // ...set $this->amqpConnection, etc
    }

    public function consume() {
        $this->amqpChannel->basic_consume('my-queue', 'my-tag', false, false, false, false, function ($msg) {
               $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        });
        
        while (count($this->amqpChannel->callbacks)) {
            try {
                $this->amqpChannel->wait();
            } catch (\Exception $e) {
                // Suppose $e is \ErrorException with "errno=32" or "Broken pipe" in message
                $this->amqpConnection->reconnect();

                 // through all of this, $this->amqpConnection->isConnected() remains true
                 // Errors with PhpAmqpLib\Exception\AMQPProtocolConnectionException: CHANNEL_ERROR - expected 'channel.open'
                $this->consume();
            }
        }
    }
}

I've tried a number of things and can't seem to get my AMQPChannel::wait() to resume. I'd be happy to add it to the examples if anyone can clarify.

RobbieLePommie · 2016-11-17T06:48:37Z

It's not a good example to keep calling a function from within itself. You'll end up with a stack overflow error.

If you write a full example then there are actually many things to take into consideration:

handling specific exceptions (e.g. a timeout may be a valid, and handleable exception)
graceful retries: depending on the error you may want to try to reconnect immediately, after 5 seconds, after 30 seconds or just give up completely.
different responses depending if you can't connect in the first place, or if the queue dies while you're looping etc.

The following is a heavily edited (simplified) version of what I'm using, but should give you an idea.

Essentially there's an outer loop that continually tried to get a connection until something in the code says "Connection Not Longer Required". If the connection dies then trap the SPECIFIC exception and handle it.

try {

    $connectionRequired = true;

    $connectionAttempts = 0;

    // This outer loop enabled the script to try to reconnect in case of failure
    while ($connectionRequired && $connectionAttempts < 30) {
        $connectionAttempts++;
        $foundQueue = false;
        try {
            // Attempt Connection . "\n";
            $connection = new AMQPStreamConnection(RABBITMQ_SERVER, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASS);
            $foundQueue = false;

            try {
                $channel = $connection->channel();
                $channel->basic_qos(null, MAX_MESSAGES_TO_HANDLE, null);
                $channel->queue_declare('task_queue', false, false, false, false); 

                $callback = function($msg){
                    // Do whatever is needed
                };

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

                // We're good!
                $foundQueue = true;

                // Reset the connection attempt counter
                $connectionAttempts = 0;

            } catch(exception $e) {
                // Failed to get channel
                // Best practice is to catch the specific exceptions and handle accordingly.
                // Either handle the message (and exit) or retry

                if (YouWantToRetry) {
                    sleep(5);  // Time should greacefully decrade based on "connectionAttempts"
                } elseif (YouCanHandleTheErrorAndWantToExitGraceully, e.g. $connectionAttempts > threshold OR youKnowTheException) {
                    $connectionRequired = false;
                } elseif (YouCannotHandleTheErrorAndWantToGetOutOfHere) {
                    throw ($e);
                }
            }
        } catch(exception $e) {
            // Failed to get connection. 
            // Best practice is to catch the specific exceptions and handle accordingly.
            // Either handle the message (and exit) or retry

            if (YouWantToRetry) {
                sleep(5);  // Time should greacefully decrade based on "connectionAttempts"
            } elseif (YouCanHandleTheErrorAndWantToExitGraceully, e.g. $connectionAttempts > threshold OR youKnowTheException) {
                $connectionRequired = false;
            } elseif (YouCannotHandleTheErrorAndWantToGetOutOfHere) {
                throw ($e);
            }

        }

        if ($foundQueue) {
            try {
                while(count($channel->callbacks)) {
                    $channel->wait(null, true, null);
                }

            } catch(exception $e) {

                // Consider this carefully! 
                // Best practice is to catch the specific exceptions and handle accordingly.

                if (YouWantToRetry) {
                    sleep(5);
                    $foundQueue = false;
                } elseif (YouCanHandleTheErrorAndWantToExitGraceully) {
                    $foundQueue = false;
                    $connectionRequired = false;
                } elseif (YouCannotHandleTheErrorAndWantToGetOutOfHere) {
                    // Error, so throw out of here
                    throw $e;
                }
            }
        }
    }

    // You'll end here on a graceful exit

} catch (Exception $e) {
    // You'll end up here if something's gone wrong

}

lukebakken · 2018-01-16T14:34:06Z

As of 2.7.0 heartbeat checks and reconnections are supported by the Stream and Socket interfaces.

Whoweez · 2018-08-24T11:57:49Z

@RobbieLePommie Hey man, thanks so much for this :) This really helped a lot!

Added link to #444

Added link to php-amqplib#444

boryashkin mentioned this issue Oct 16, 2017

check_heartbeat in write() at StreamIO #507

Closed

lukebakken closed this as completed Jan 16, 2018

Maxim-Mazurok added a commit to Maxim-Mazurok/php-amqplib that referenced this issue Jul 6, 2019

Added link to php-amqplib#444

725dd61

lukebakken added a commit that referenced this issue Jul 8, 2019

Merge pull request #709 from Maxim-Mazurok/patch-1

5f4afed

Added link to #444

kratkyzobak pushed a commit to kratkyzobak/php-amqplib that referenced this issue Feb 9, 2024

Added link to php-amqplib#444

5f89c52

kratkyzobak pushed a commit to kratkyzobak/php-amqplib that referenced this issue Feb 9, 2024

Merge pull request php-amqplib#709 from Maxim-Mazurok/patch-1

7c81f6c

Added link to php-amqplib#444

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Reconnection guidelines - graceful dropped connection handling on consumer #444

Reconnection guidelines - graceful dropped connection handling on consumer #444

Uh oh!

Uh oh!

Uh oh!

Reconnection guidelines - graceful dropped connection handling on consumer #444

Reconnection guidelines - graceful dropped connection handling on consumer #444

Comments

Uh oh!

Uh oh!

Uh oh!

Uh oh!