Как использовать RabbitMQ с PHP

 <?php chdir(dirname(__DIR__)); require_once('vendor/autoload.php'); use Acme\AmqpWrapper\SimpleSender; $theName = filter_input(INPUT_POST, 'theName', FILTER_SANITIZE_STRING); $simpleSender = new SimpleSender(); $simpleSender->execute($theName); header("Location: orderReceived.html"); 

Этот код просто проверит параметр POST с именем theName и отправит его объекту, который мы создали для его обработки. Давайте посмотрим на метод SimpleSender :: execute ():

 <?php // ... SOME CODE HERE ... /** * Sends a message to the pizzaTime queue. * * @param string $message */ public function execute($message) { /** * Create a connection to RabbitAMQP */ $connection = new AMQPConnection( 'localhost', #host - host name where the RabbitMQ server is runing 5672, #port - port number of the service, 5672 is the default 'guest', #user - username to connect to server 'guest' #password ); /** @var $channel AMQPChannel */ $channel = $connection->channel(); $channel->queue_declare( 'pizzaTime', #queue name - Queue names may be up to 255 bytes of UTF-8 characters false, #passive - can use this to check whether an exchange exists without modifying the server state false, #durable - make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart false, #exclusive - used by only one connection and the queue will be deleted when that connection closes false #autodelete - queue is deleted when last consumer unsubscribes ); $msg = new AMQPMessage($message); $channel->basic_publish( $msg, #message '', #exchange 'pizzaTime' #routing key ); $channel->close(); $connection->close(); } 

Строковая разбивка выглядит следующим образом:

 <?php /* ... MORE CODE HERE ... */ $connection = new AMQPConnection( 'localhost', #host - host name where the RabbitMQ server is runing 5672, #port - port number of the service, 5672 is the default 'guest', #user - username to connect to server 'guest' #password ); /* ... MORE CODE HERE ... */ 

Сначала мы создаем объект подключения. Обратите внимание, что учетные данные guest: guest являются значениями по умолчанию для RabbitMQ. Тем не менее, вам будет разрешено подключаться к серверу только с его помощью, если вы подключаетесь с одного хоста (localhost).

Поскольку RabbitMQ прослушивает и обслуживает один порт, нам нужно создать канал (представьте его как виртуальный порт) с помощью $channel = $connection->channel(); поэтому другие клиенты могут подключаться к серверу.

 <?php /* ... MORE CODE HERE ... */ $channel->queue_declare( 'pizzaTime', #queue name - Queue names may be up to 255 bytes of UTF-8 characters false, #passive - can use this to check whether an exchange exists without modifying the server state false, #durable - make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart false, #exclusive - used by only one connection and the queue will be deleted when that connection closes false #autodelete - queue is deleted when last consumer unsubscribes ); /* ... MORE CODE HERE ... */ 

Как только у нас будет готов канал, давайте объявим очередь для отправки запроса. Хорошая вещь о RabbitMQ заключается в том, что мы можем создавать очереди непосредственно из клиента, но мы должны быть осторожны при его создании. Давайте кратко объясним параметры, используемые для создания очереди с помощью $channel->queue_declare()

• Имя очереди : это произвольное имя, будет использоваться для идентификации очереди
• Пассивный : если установлено значение true, сервер будет проверять только, может ли быть создана очередь, false будет фактически пытаться создать очередь.
• Долговечный : как правило, если сервер останавливается или аварийно завершает работу, все очереди и сообщения теряются … если только мы не объявим очередь длительной, и в этом случае очередь сохранится, если сервер будет перезапущен.
• Исключительно : если true, очередь может использоваться только тем соединением, которое ее создало.
• Автоудаление : при значении true очередь будет удалена, если в ней нет сообщений и нет подключенных подписчиков.

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

Затем мы создали объект сообщения с $msg = new AMQPMessage($message); $message — параметр POST, theName которое мы получили из формы. Сообщение может быть любой строкой.

 <?php /* ... MORE CODE HERE ... */ $channel->basic_publish( $msg, #message '', #exchange 'pizzaTime' #routing key ); /* ... MORE CODE HERE ... */ 

Теперь мы должны опубликовать сообщение в очереди. Однако мы не можем публиковать сообщения напрямую в очередь, если это не происходит через обмен. Мы никогда не объявляли обмен, так как это будет возможно? Оказывается, что когда мы создаем очередь без определения обмена, к которому будет привязана очередь, будет использоваться обмен по умолчанию. Мы можем опубликовать сообщение в очереди через обмен по умолчанию с помощью $channel->basic_publish() , параметры которого используются:

• Сообщение: сообщение, которое мы хотим отправить
• Exchange: обратите внимание, что мы используем пустую строку, потому что мы будем использовать обмен по умолчанию
• Ключ маршрутизации: имя очереди, в которую мы хотим доставить сообщение.

 <?php /* ... MORE CODE HERE ... */ $channel->close(); $connection->close(); /* ... MORE CODE HERE ... */ 

После того, как мы закончим, мы должны закрыть соединение с каналом и сервером.

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

Итак, заказ пиццы находится в очереди, как мы их получим? Прежде всего, мы должны знать, что потребитель должен установить постоянное соединение с сервером очередей (или подписаться), чтобы он мог получать сообщения от сервера. Сервер не будет отправлять эти сообщения в наше приложение. К счастью, создать такое соединение очень просто.

 <?php namespace Acme\AmqpWrapper; use PhpAmqpLib\Connection\AMQPConnection; class SimpleReceiver { /* ... SOME CODE HERE ... */ /** * Listens for incoming messages */ public function listen() { $connection = new AMQPConnection( 'localhost', #host 5672, #port 'guest', #user 'guest' #password ); $channel = $connection->channel(); $channel->queue_declare( 'pizzaTime', #queue name, the same as the sender false, #passive false, #durable false, #exclusive false #autodelete ); $channel->basic_consume( 'pizzaTime', #queue '', #consumer tag - Identifier for the consumer, valid within the current channel. just string false, #no local - TRUE: the server will not send messages to the connection that published them true, #no ack - send a proper acknowledgment from the worker, once we're done with a task false, #exclusive - queues may only be accessed by the current connection false, #no wait - TRUE: the server will not respond to the method. The client should not wait for a reply method array($this, 'processOrder') #callback - method that will receive the message ); while(count($channel->callbacks)) { $channel->wait(); } $channel->close(); $connection->close(); } /** * @param $msg */ public function processOrder($msg) { /* ... CODE TO PROCESS ORDER HERE ... */ } } 

Так же, как мы подключились, создали канал и объявили очередь в производителе, мы должны сделать то же самое внутри потребителя. Однако в потребителе мы должны подписаться на канал с помощью $channel->basic_consume() , и используемые параметры определяются следующим образом:

• Очередь: должно быть таким же именем очереди, которое мы определили в производителе
• Потребительский тег: произвольное имя, данное потребителю. Если это поле не заполнено, сервер сгенерирует уникальный тег
• Нет локальных: это скрытый параметр, при активации сервер не будет доставлять свои собственные сообщения
• No Ack (nowledgement): автоматически подтвердит, что потребитель получил сообщение, поэтому нам не нужно делать это вручную.
• Нет ожидания: если установлено, сервер не будет ждать завершения процесса в получателе
• Обратный вызов: может быть именем функции, массивом, содержащим объект и имя метода, или замыканием, которое получит сообщение в очереди. Этот обратный вызов должен принимать параметр, содержащий такое сообщение. В нашем примере array($this, 'processOrder') используется для определения метода processOrder() текущего объекта в качестве обратного вызова.

 <?php /* ... SOME CODE HERE ... */ while(count($channel->callbacks)) { $channel->wait(); } /* ... SOME CODE HERE ... */ 

«Волшебство» происходит внутри цикла while. Если у подписчика есть хотя бы один определенный обратный вызов, мы будем ждать любого события в канале. Каждый раз, когда поступает сообщение, выполняется наш определенный обратный вызов processOrder() , поэтому мы можем обрабатывать сообщение по мере необходимости.

Как мы можем запустить это? Просто создайте сценарий, который будет вызывать метод SimpleReceiver::listen() например:

 <?php chdir(dirname(__DIR__)); require_once('vendor/autoload.php'); use Acme\AmqpWrapper\SimpleReceiver; $receiver = new SimpleReceiver(); $receiver->listen(); 

Теперь запустите процесс в консоли с помощью php <script name> и дайте ему выполнить свою работу. Если вы хотите убить потребителя, простой Ctrl + C прервет процесс.

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

Вывод

В этой части мы представили теорию AMQP и системы массового обслуживания и продемонстрировали их использование на простом примере. В продолжение этого поста мы рассмотрим еще два примера повышенной сложности и сложных концепций. Будьте на связи!