Laravel & RabbitMQ – PHP Tutoriais

Usando RabbitMQ no Laravel

Neste artigo vamos explicar como usar um Message Broker e como criar um setaup de um exemplo com RabbitMQ e Laravel

O que é RabbitMQ e qual sua função na programação?

RabbitMQ é um software de mensageria de código aberto que permite que aplicativos se comuniquem uns com os outros usando filas de mensagens através do protocolo Advanced Message Queuing Protocol (AMQP).

Além do protocolo AMQP, o RabbitMQ também suporta os protocolos:

HTTP
STOMP
MQTT

Quando usar um Message Broker ?

Para exemplificar vamos pegar uma cafeteria como um exemplo do mundo real. Quando clientes fazem um pedido, em geral tem alguém que vai anotar seu pedido. Quando um cliente faz um pedido ele não para de receber mais pedidos, ele retorna com uma mensagem dizendo “Sente-se ali que seu pedido será preparado em breve”.

Agora o atendente coloca os diferentes pedidos feitos pelos clientes em uma lista enquanto ele continua a anotar novos pedidos. A lista pode ser composta, por exemplo, do numero do pedido e seus itens. Veja um exemplo abaixo:

1, Expresso

2, Cappuccino

3, Café com Leite

4, Expresso

5, Pão na Chapa e Café com leite

Se o atendente tivesse que parar de receber novos pedidos porque precisava fazer o primeiro pedido, haveria um tempo de espera desnecessário que poderia impactar todo o negócio. Então enquanto alguém faz o café, o atendente pode receber outros pedidos.

Desta forma quando um pedido é finalizado e entregue na mesa do cliente o pedido deve ser removido da lista, que nos termos do Message Broker significa remover da fila (queue). O pedido só sai de fato da fila quando é entregue ao cliente, de forma que o cliente não ficou esperando uma resposta (síncrono), ele foi notificado quando ficou pronto (assíncrono).

Durante o período em que seu pedido estava sendo preparado, o cliente estava disponível para realizar outras tarefas.

Eles podem olhar o celular, fazer novos pedidos, responder um e-mail, etc. Em termos técnicos, permitimos que o cliente ficasse mais feliz em distribuir seus recursos em outro lugar, em vez de apenas esperar por uma resposta.

Agora imagine que existe um tipo de café que leva menos tempo para preparar (Café com leite) e existe outro tipo que leva mais tempo para preparar (Cappuccino), então esses pedidos podem ser feitos de acordo com a prioridade e permitir que os clientes gastem o tempo com mais sabedoria apenas usando processamento assíncrono.

Agora, o que poderia acontecer a seguir para o negócio de cafeterias? Pode crescer e começar a franquear e ter uma rede maior.

Digamos algo como Starbucks. Supondo que existe Starbucks 1, Starbucks 2, Starbucks 3, etc. E cada filial atende aproximadamente milhares de clientes.

Agora, considerando o pior dia de uma empresa onde há uma queda de energia em uma cidade onde está localizada uma filial do Starbucks 1. Como podemos atender os pedidos que já foram recebidos? A filial mais próxima recebe esses pedidos e os prepara e então os clientes serão notificados quando estiverem prontos para serem retirados.

Mas como você pode fazer isso tecnicamente?

Bem, a maneira simples de manter uma lista na memória não funcionará porque, quando a loja estiver fora do ar, ela perderá eletricidade e o computador será desligado, então você precisará de algum tipo de persistência em seus dados.

Digamos que tentamos fazer isso com um banco de dados onde a lista é armazenada em um servidor e os balanceadores de carga são configurados com 4 nós chamados n1, n2, n3, n4.

O que acontece se um nó cair, digamos, por exemplo, n2?

Podemos acompanhar quais pedidos são atendidos por quais nós no banco de dados e tentar atender por meio de nós disponíveis, mas isso fica complicado quando mais solicitações e mais nós caem, mantendo também o status dos pedidos e a priorização.

De uma perspectiva do mundo real, os seguintes problemas podem acontecer, o café errado pode ser entregue ao cliente errado ou o mesmo cliente pode receber vários cafés ou o cliente pode não receber nenhum café.

Isso pode ser resolvido através de balanceamento de carga e algum tipo de mecanismo de pulsação, você pode notificar todos os pedidos com falha aos servidores disponíveis.

Agora, e se você quiser todos os recursos de notificação, balanceamento de carga, pulsação e persistência em uma só coisa?

Isso seria um Message Queue.

O que ele faz é pegar tarefas, persisti-las, atribuí-las ao servidor correto e esperar que sejam concluídas. Se demorar muito para o servidor confirmar, ele percebe que o servidor está inoperante e então o atribui ao próximo servidor. Uma fila de mensagens tem várias estratégias de como isso pode ser alcançado, mas isso está encapsulado no message queue, por exemplo, RabbitMQ.

Existem algumas alternativas ao RabbitMQ, que podem ser:

IBM MQ.
Apache Kafka.
Google Cloud Pub/Sub.
Amazon MQ.
Apache ActiveMQ.
KubeMQ.
ZeroMQ.

Como criar uma instância do RabbitMQ

Em um arquivo docker compose, inclua a seção abaixo e em seguida execute docker compose up -d.

services:
  # MessageBroker
  rabbit:
    image: rabbitmq:3-management
    container_name: rabbitmq-server
    ports:
      - 15672:15672
      - 5672:5672
    networks:
      - internal
    volumes:
      - rabbitmq-conf:/etc/rabbitmq/rabbitmq.conf
      - rabbitmq-data:/var/lib/rabbitmq
networks:
  internal:
    driver: bridge

volumes:
  rabbitmq-conf:    
  rabbitmq-data:

Isso vai lançar um container docker do RabbitMQ que inclui um site de gerenciamento. Você pode acessar pelo endereço localhost:15672

Instalando o RabbitMQ no Laravel

O Laravel não suporta o RabbitMQ por padrão, mas existe um pacote que pode habilitar este suporte. Podemos usar o composer para instalar o pacote ao nosso projeto. https://github.com/vyuldashev/laravel-queue-rabbitmq

$ composer require vladimir-yuldashev/laravel-queue-rabbitmq

E para criar a conexão do PHP com o RabbitMQ, você deve instalar algum pacote que realize a comunicação via protocolo AMQP

$ composer require php-amqplib/php-amqplib
$ composer require enqueue/amqp-lib

Agora é necessário realizar a configuração dentro do projeto Laravel. Edite o arquivo config/queue.php e inclua a configuração abaixo para novo driver de queue.

    'connections' => [
     ...
       'rabbitmq' => [
            'driver' => 'rabbitmq',/*
            * Set to "horizon" if you wish to use Laravel Horizon.
            */
            'worker' => env('RABBITMQ_WORKER', 'default'),
            'dsn' => env('RABBITMQ_DSN', null),
            'factory_class' => AmqpConnectionFactory::class,
            'host' => env('RABBITMQ_HOST', 'rabbitmq'),
            'port' => env('RABBITMQ_PORT', 5672),
            'vhost' => env('RABBITMQ_VHOST', '/'),
            'login' => env('RABBITMQ_LOGIN', 'guest'),
            'password' => env('RABBITMQ_PASSWORD', 'guest'),
            'queue' => env('RABBITMQ_QUEUE', 'default'),
            'options' => [
                'exchange' => [
                    'name' => env('RABBITMQ_EXCHANGE_NAME'),/*
                    * Determine if exchange should be created if it does not exist.
                    */
                    'declare' => env('RABBITMQ_EXCHANGE_DECLARE', true),/*
                    * Read more about possible values at https://www.rabbitmq.com/tutorials/amqp-concepts.html
                    */'type' => env('RABBITMQ_EXCHANGE_TYPE',\Interop\Amqp\AmqpTopic::TYPE_DIRECT),
                    'passive' => env('RABBITMQ_EXCHANGE_PASSIVE', false),
                    'durable' => env('RABBITMQ_EXCHANGE_DURABLE', true),
                    'auto_delete' => env('RABBITMQ_EXCHANGE_AUTODELETE', false),
                    'arguments' => env('RABBITMQ_EXCHANGE_ARGUMENTS'),
                ],
                'queue' => [
                    /*
                    * Determine if queue should be created if it does not exist.
                    */'declare' => env('RABBITMQ_QUEUE_DECLARE', true),
                    /*
                    * Determine if queue should be binded to the exchange created.
                    */'bind' => env('RABBITMQ_QUEUE_DECLARE_BIND', true),
                    /*
                    * Read more about possible values at https://www.rabbitmq.com/tutorials/amqp-concepts.html
                    */'passive' => env('RABBITMQ_QUEUE_PASSIVE', false),
                    'durable' => env('RABBITMQ_QUEUE_DURABLE', true),
                    'exclusive' => env('RABBITMQ_QUEUE_EXCLUSIVE', false),
                    'auto_delete' => env('RABBITMQ_QUEUE_AUTODELETE', false),
                    'arguments' => env('RABBITMQ_QUEUE_ARGUMENTS'),
                ],
            ],/** Determine the number of seconds to sleep if there's an error communicating with rabbitmq
            * If set to false, it'll throw an exception rather than doing the sleep for X seconds.
            */
            'sleep_on_error' => env('RABBITMQ_ERROR_SLEEP', 5),
            /*
            * Optional SSL params if an SSL connection is used
            * Using an SSL connection will also require to configure your RabbitMQ to enable SSL. More details can be founds here: https://www.rabbitmq.com/ssl.html
            */'ssl_params' => [
            'ssl_on' => env('RABBITMQ_SSL', false),
            'cafile' => env('RABBITMQ_SSL_CAFILE', null),
            'local_cert' => env('RABBITMQ_SSL_LOCALCERT', null),
            'local_key' => env('RABBITMQ_SSL_LOCALKEY', null),
            'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
            'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
            ],
        ],
    ],
...

Dentro do seu arquivo .env adicione as seguintes entradas:

QUEUE_CONNECTION=rabbitmq
RABBITMQ_HOST=rabbitmq
RABBITMQ_PORT=5672
RABBITMQ_VHOST=/
RABBITMQ_LOGIN=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_QUEUE=defaultRABBITMQ_EXCHANGE_NAME=default
RABBITMQ_EXCHANGE_DECLARE=
RABBITMQ_EXCHANGE_TYPE=
RABBITMQ_EXCHANGE_PASSIVE=
RABBITMQ_EXCHANGE_DURABLE=
RABBITMQ_EXCHANGE_AUTODELETE=
RABBITMQ_EXCHANGE_ARGUMENTS=defaultRABBITMQ_QUEUE_DECLARE=
RABBITMQ_QUEUE_DECLARE_BIND=
RABBITMQ_QUEUE_PASSIVE=
RABBITMQ_QUEUE_DURABLE=
RABBITMQ_QUEUE_EXCLUSIVE=
RABBITMQ_QUEUE_AUTODELETE=
RABBITMQ_QUEUE_ARGUMENTS=defaultRABBITMQ_ERROR_SLEEP=5
RABBITMQ_SSL=
RABBITMQ_SSL_CAFILE=
RABBITMQ_SSL_LOCALCERT=
RABBITMQ_SSL_LOCALKEY=
RABBITMQ_SSL_VERIFY_PEER=
RABBITMQ_SSL_PASSPHRASE=

Agora podemos criar um Job para realizar um teste de ver como processo funciona

php artisan make:job TestQueue

Em seguida podemos criar um command para disparar a execução do Job

php artisan make:command TestCommand

TestQueue.php

<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;

class TestQueue implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public array $data;
    /**
     * Create a new job instance.
     */
    public function __construct($data)
    {
        //
        $this->data = $data;
    }

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        //
        $data = $this->data;
        Log::info($data);
    }
}

TestCommand.php

<?php

namespace App\Console\Commands;

use App\Jobs\TestQueue;
use Illuminate\Console\Command;

class TestCommand extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'app:test-command';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Command description';

    /**
     * Execute the console command.
     */
    public function handle()
    {
        //
        $data = ['name' => 'PHP Tutoriais', 'phone' => '+551198765321'];
        TestQueue::dispatch($data);
    }
}

Para executar a fila é necessário executar comando do artisan