user2016 user2016 - 2 months ago 30
PHP Question

RabbitMQ & php-amqplib: cunsumer reconnect

I have a consumer worker in PHP script.

But sometimes the RabbitMQ server stops from running,

I get this error:

PHP Fatal error: Uncaught exception 'ErrorException' with message 'fwrite(): send of 19 bytes failed with errno=32 Broken pipe' in /home/user/pusher/rabbitmq-worker/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Wire/IO/StreamIO.php:281


I would like to be able to handle this error with a try/catch block to throw a nice error to my console or try to reconnect after a while.

This is what I have code so far:

while (true) {
try {
$connection = new AMQPStreamConnection(RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASS);
$channel = $connection->channel();
$channel->queue_declare(RABBITMQ_DT_QUEUE, false, true, false, false);
$channel->basic_qos(null, 11, null);
echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($req) {
sleep(3);//Some task that takes 3 sec
};
$channel->basic_consume(RABBITMQ_QUEUE, '', false, false, false, false, $callback);
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
} catch (Exception $e) {
sleep(1);
}


What's missing?

Answer

Define next parameter on connection

keepalive: true
heartbeat: 15 // An example

It is two latest arguments of AMQPStreamConnection::__constructor()

NOTE: if you connect to some distant server, I recommend set high up timeouts

connection_timeout: 10
read_write_timeout: 30

UPD: You can catch ErrorException and try connect to server again