RabbitMQ + PHP quick start

1 Установка RabbitMQ

Сам RMQ на Убунту ставится так:

echo 'deb http://www.rabbitmq.com/debian/ testing main' | sudo tee /etc/apt/sources.list.d/rabbitmq.list
wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add -
sudo apt-get update
sudo apt-get install rabbitmq-server
sudo rabbitmq-plugins enable rabbitmq_management

Теперь если открыть ваш localhost:15672 под учеткой: guest guest и увидеть много прикольных вещей.

По умолчанию гостевой пользователь существует и может подключаться только с localhost.
Создание пользователя с правами админа с доступом отовсюду выглядит так:

sudo rabbitmqctl add_user admin StrongPassword
sudo rabbitmqctl set_user_tags admin administrator

По умолчанию новый пользователь не имеет доступа к Virtual Hosts.
Вам следует войти в админку, там на вкладке "Admin" нажать кнопку "Ser Permission" для нового пользователя.

Важная инфа:

• Порт 15672 - для Веб интерфейса rabbitmq
• Порт 5672 - для подключения клиента rabbitmq

2 Пример RabbitMQ на PHP

Пример будет показан на консольном приложении где один процесс будет рассылать сообщения, а другой принимать.

Установим RMQ-PHP библиотеку в корне проекта примера, например /var/www/rmq:

composer require php-amqplib/php-amqplib

• Код фаила конфигурации config.php

<?php

require_once __DIR__ . '/vendor/autoload.php';

putenv('TEST_AMQP_DEBUG=1');
require_once __DIR__ . '/config.php';

const HOST = 'localhost';
const PORT = '5672';
const USER = 'admin';
const PASS = 'StrongPassword';
const VHOST = '/';

Нужно будет открыть 2 вкладки в терминале, условимся называть одну - Издатель, другую - Подписчик

• Код Издателя amqp_publisher.php

<?php

include(__DIR__ . '/config.php');

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

$exchange = 'router';
$queue = 'msgs';

$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);
$channel = $connection->channel();

/*
    The following code is the same both in the consumer and the producer.
    In this way we are sure we always have a queue to consume from and an
        exchange where to publish messages.
*/

/*
    name: $queue
    passive: false
    durable: true // the queue will survive server restarts
    exclusive: false // the queue can be accessed in other channels
    auto_delete: false //the queue won't be deleted once the channel is closed.
*/
$channel->queue_declare($queue, false, true, false, false);

/*
    name: $exchange
    type: direct
    passive: false
    durable: true // the exchange will survive server restarts
    auto_delete: false //the exchange won't be deleted once the channel is closed.
*/

$channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);

$channel->queue_bind($queue, $exchange);

$messageBody = implode(' ', array_slice($argv, 1));
$message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
$channel->basic_publish($message, $exchange);

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

• Код Подписчика amqp_consumer.php

<?php

include(__DIR__ . '/config.php');
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;

$exchange = 'router';
$queue = 'msgs';
$consumerTag = 'consumer';

$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);
$channel = $connection->channel();

/*
    The following code is the same both in the consumer and the producer.
    In this way we are sure we always have a queue to consume from and an
        exchange where to publish messages.
*/

/*
    name: $queue
    passive: false
    durable: true // the queue will survive server restarts
    exclusive: false // the queue can be accessed in other channels
    auto_delete: false //the queue won't be deleted once the channel is closed.
*/
$channel->queue_declare($queue, false, true, false, false);

/*
    name: $exchange
    type: direct
    passive: false
    durable: true // the exchange will survive server restarts
    auto_delete: false //the exchange won't be deleted once the channel is closed.
*/

$channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);

$channel->queue_bind($queue, $exchange);

/**
 * @param \PhpAmqpLib\Message\AMQPMessage $message
 */
function process_message($message)
{
    echo "\n--------\n";
    echo $message->body;
    echo "\n--------\n";

    $message->ack();

    // Send a message with the string "quit" to cancel the consumer.
    if ($message->body === 'quit') {
        $message->getChannel()->basic_cancel($message->getConsumerTag());
    }
}

/*
    queue: Queue from where to get the messages
    consumer_tag: Consumer identifier
    no_local: Don't receive messages published by this consumer.
    no_ack: If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.
    exclusive: Request exclusive consumer access, meaning only this consumer can access the queue
    nowait:
    callback: A PHP Callback
*/

$channel->basic_consume($queue, $consumerTag, false, false, false, false, 'process_message');

/**
 * @param \PhpAmqpLib\Channel\AMQPChannel $channel
 * @param \PhpAmqpLib\Connection\AbstractConnection $connection
 */
function shutdown($channel, $connection)
{
    $channel->close();
    $connection->close();
}

register_shutdown_function('shutdown', $channel, $connection);

// Loop as long as the channel has callbacks registered
while ($channel->is_consuming()) {
    $channel->wait();
}

• Теперь во вкладке терминала "Подписчик" выполним команду

php amqp_consumer.php

Если ошибок не появится значит клиент успешно подключился к RMQ и процесс перешел в режим ожидания сообщений

• Теперь во вкладке терминала "Издатель" выполним команду

php amqp_publisher.php some text to publish

Издатель отправит сообщение на сервер RMQ который в свою очередь перешлет его всем подписчикам
Вернемся на вкладку "Подписчик" и увидим что он получил сообщение от издателя.

Источники:

Как установить RabbitMQ на Ubuntu 20.04 LTS https://infoit.com.ua/linux/kak-ustanovit-rabbitmq-na-ubuntu-20-04-lts/
Symfony + RabbitMQ Быстрый старт для молодых https://habr.com/ru/post/338950/
php-amqplib/php-amqplib https://github.com/php-amqplib/php-amqplib