testconsumer.php

<?php

require('vendor/autoload.php');

use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Connection\AMQPLazyConnection;
use PhpAmqpLib\Connection\AMQPSocketConnection;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

if (empty($argv[1])) {
    die('queue name...?');
}

$exchangeName = 'testexchange';
$queueName = $argv[1];

$conn = new AMQPStreamConnection(
    '172.24.55.150',
    5672,
    'test',
    'test',
    '/',
    false, 'AMQPLAIN', null, 'en_US',
    3, 3, null, false, 2
);

$channel = $conn->channel();
$channel->exchange_declare($exchangeName, 'fanout', false, true, false);
$channel->queue_declare($queueName, false, true, false, false);
$channel->queue_bind($queueName, $exchangeName);

$consumerTag = 'consumer:' . uniqid();

$msgHandler = new MsgHandler();

$channel->basic_consume($queueName, $consumerTag, false, false, false, false, [$msgHandler, 'processMessage']);

$i = 0;
while (true) {

    try {
        $channel->wait(null, true);
    } catch (\Exception $e) {
        echo gmdate('r') . ': wait exception : ' . $e->getMessage() . PHP_EOL;
        $conn = $channel->getConnection();
        try {
            $conn->reconnect();
            $channel = $conn->channel();
            $channel->exchange_declare($exchangeName, 'fanout', false, true, false);
            $channel->queue_declare($queueName, false, true, false, false);
            $channel->queue_bind($queueName, $exchangeName);
        } catch (\Exception $e) {
            echo gmdate('r') . ':reconnect exception : '. $e->getMessage() . PHP_EOL;
        }
    }

    echo gmdate('r') . ":received : " . $msgHandler->body . PHP_EOL;
}

class MsgHandler {
    public $body = '';
    function processMessage($msg) {
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        $this->body = $msg->body;
    }
}