PHP. Разработка первого приложения с использованием RabbitMQ

RabbitMQ является брокером сообщений: он принимает и пересылает сообщения.
Он работает по сути так-же как почтовое отделение. Вы кладете почту, которую хотите отправить в почтовый ящик.
После чего вы можете выть уверены, что сотрудник почты в конечном итоге доставит вашу почту получателю.
RabbitMQ работает аналогично данному примеру, представляя из себя почтовый ящик, почтовое отделения и почтальона.

Основное различие между RabbitMQ и почтой заключается в том, что она не имеет дело с бумагой, а принимает, хранит и пересылает двоичные данные - сообщения.

В данной статье будет опубликован перевод первого из шести уроков с официального сайта.

• Producer - 'отправитель', отправляющий сообщения. В схемах он будет представлен буквой "P", обведенной в овал:
  
  
  
  
• Queue - 'очередь'. Это имя почтового ящика, который находится внутри RabbitMQ. Хотя сообщения проходят через RabbitMQ и ваше приложение, они могут храниться только внутри очереди. Очередь не имеет ограничений на количество принимаемых сообщений. По сути можно считать, что буфер является бесконечным и ограничен только ресурсами машины.
  Любое количество отправителей может отправлять сообщения в очередь, а любое количество получателей может попытаться получить данные из этой очереди.
  В схемах очередь будет изображена следующим образом:
  
  
  
  
• Consumer - 'получатель', принимающий сообщения. Обычно получатель находится в состоянии ожидания сообщений.
  В схемах он будет представлен буквой "C", обведенной в овал:
  
  
  
  Брокер, отправитель и получатель не обязаны находиться на одной и той же машине. Обычно они находятся на разных машинах.
  

"Hello World"

В этой части урока мы напишем две программы на PHP: отправитель, который отправляет одно сообщение, и получатель, который получает сообщения и распечатывает их. Для разработки нам понадобится скачать и установить расширение php-amqplib.

Скачивание и установка php-amqplib

На приведенной ниже диаграмме "P" - наш отправитель, а "C " - наш получатель. Поле, изображенное в центре - это очередь-буфер сообщений, который RabbitMQ хранит с именем получатель.

Теперь, когда библиотека php-amqplib установлена, мы можем приступить к написанию кода.

Sending ('Отправитель').

Мы создадим файл send.php и напишем в нем код, отвечающий за отправку сообщений и файл receie.php, отвечающий за получение сообщений. Отправитель подключится к RabbitMQ, отправит одно сообщение, а затем выйдет.

В файле send.php, нам нужно подключить библиотеку и использовать необходимые классы:

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

Затем мы можем создать соединение с сервером:

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

Соединение абстрагирует соединение сокета и заботится о согласовании версии протокола и аутентификации и так далее. Здесь мы подключаемся к брокеру на локальной машине - localhost. Если мы хотим подключиться к брокеру на другую машину, то должны указать доменное имя машины или IP-адрес брокера.
Затем мы создаем канал, в котором находится большая часть API для выполнения задач.

Для того, чтобы отправить сообщение, мы должны объявить очередь для отправки, затем мы можем опубликовать сообщение в очереди:

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

$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'hello');

echo " [x] Sent 'Hello World!'\n";

Объявление очереди является идемпотентным - она будет создана, только если она еще не существует. Содержимое сообщения представляет собой массив байтов, поэтому вы можете кодировать все, что вам нравится.

После чего, мы закрываем канал и соединение:

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

В итоге файл send.php будет выглядеть следующим образом:

Receiving ('Получатель').

Приемник прослушивает сообщения от RabbitMQ, поэтому, в отличие от издателя, который публикует одно сообщение, мы будем продолжать прослушивать и распечатывать весь список сообщений.


Файл receive.php начинается с такого же кода, как в файле send.php:

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

Настройка совпадает с настройкой отправителя: мы открываем соединение и канал и объявляем очередь, из которой собираемся забирать сообщения. Обратите внимание, что этот код совпадает с кодом файла send.php.

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

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

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

Обратите внимание, что мы объявляем очередь и здесь. Поскольку можем запустить получателя раньше отправителя. Убедимся, что очередь существует, прежде чем попытаемся использовать сообщения из нее.

Мы хотим сообщить серверу, что он должен доставить нам сообщения из очереди.
Определим PHP callback функцию, которая будет получать сообщения отправленные сервером.
Имейте в виду, что сообщения отправляются асинхронно с сервера на клиент.

$callback = function ($msg) {
  echo ' [x] Received ', $msg->body, "\n";
};

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

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

Выполнение кода будет заблокировано до тех пор, пока $channel имеет обратные вызовы.
Всякий раз, когда мы получаем сообщение, наша callback функция принимает его и позволяет с ним работать.

Полный, готовый код файла receive.php:

Собрать все воедино.

Теперь мы можем запустить оба сценария. В терминале запустите receive (Получатель):

php receive.php

затем запустите send (Отправитель):

php send.php

Получатель напечатает сообщение, которое он получает от отправителя через RabbitMQ. Получатель будет продолжать работать, ожидая сообщения (используйте Ctrl-C, чтобы остановить его), поэтому попробуйте запустить отправителя с другого терминала.