RabbitMQ + Laravel

Vou abordar nesse artigo alguns conceitos de Fila e Mensageria com RabbitMQ, mas para isso vou partir do pré-suposto que você já conhece um pouco sobre o Laravel, Menssageria, Filas e também Docker.

Antes de iniciar e pormos a mão na massa, precisamos entender como funciona o RabbitMQ com o fluxo de uma mensagem.

1- Producer

O Producer é aplicação ou serviço que irá realizar o envio de uma mensagem;

2- Exchanges

A Exchanges é para onde o Producer envia a mensagem. No RabbitMQ a aplicação nunca envia a mensagem direto para fila, sempre para uma exchange e ela faz o redirecionamento para a fila específica. Uma exchange pode possuir varias filas(queues), e o ideal é que uma fila só tenha uma exchange.

3- Binding

O Binding é uma ligação ou um link entre uma fila e uma exchange.

4- Queues

As Queues são as filas onde as mensagens serão armazenadas.

5- Consumer

O Consumer é a aplicação que irá pegar a mensagem na fila e realizar seu processamento.

Tipos de Exchanges

Existem vários tipos de exchanges, e cada tipo tem uma finalidade na hora de sua concepção. A Direct, a Topic e a Fanout estão representadas na imagem abaixo. Em nosso projeto vamos utilizar apenas o tipo Direct.

Direct: A mensagem é roteada para as filas(queues) cuja chave de ligação corresponde exatamente à chave de roteamento da mensagem. Por exemplo, se a fila estiver vinculada à exchange com um binding key pdf_process, uma mensagem publicada na exchange com uma routing key pdf_process será roteada para essa fila(queue).
Fanout: O tipo de exchange fanout roteia mensagens para todas as filas vinculadas a ele.
Topic: O tipo de exchange topic, faz uma correspondência curinga entre a chave de roteamento e o padrão de roteamento especificado na ligação.
Headers: as trocas de exchanges usam os atributos do cabeçalho da mensagem para roteamento.

Criando o Projeto

Vamos iniciar instalando um projeto Laravel 10.x (link).
Depois disso você precisa instalar o Laravel Sail (link).

Agora só nos falta criar um contêiner com o RabbitMQ para nosso projeto.

Na raiz do projeto abra em seu editor de códigos(vscode) o arquivo docker-compose.yml, criado pelo Laravel-Sail, e adicione o código abaixo, mais informações sobre as imagens Dockerhub.

rabbitmq:
        image: rabbitmq:3-management
        container_name: 'rabbitmq'
        ports:
            - 5672:5672
            - 15672:15672
        volumes:
            - ~/.docker/rabbitmq/data/:/var/lib/rabbitmq/
            - ~/.docker/rabbitmq/log/:/var/log/rabbitmq
        networks:
            - sail

Basta startar os contêineres com o Laravel Sail e pronto o RabbitMQ estará funcionando, sendo agora possível testar em http://localhost:15672

Usuário: guest
Senha: guest

Configurando o RabbitMQ

Vou aqui já definir as nomenclaturas do que vamos configurar no RabbitMQ, ficando assim o mais objetivo e visual o possível.

Cenário 1

Exchange: pdf_events
Queue A: create_pdf_queue
Binding entre a exchange (pdf_events) e a Queue A (create_pdf_queue): pdf_create

Cenário 2

Exchange: pdf_events
Queue B: pdf_log_queue
Binding entre a exchange (pdf_events) e a Queue B (pdf_log_queue): pdf_log

Queues do RabbitMQ

Primeiro vamos criar as Queues, elas serão do tipo Classic, com os nomes create_pdf_queue e pdf_log_queue.

Exchange do RabbitMQ

Agora vamos criar a exchange, que vai ser do tipo direct e Durable, com o nome pdf_events.

Bindings do RabbitMQ

Acesse as Queues, abra a fila create_pdf_queue e vamos criar um binding entre a fila e a exchange pdf_events, ao qual vamos dar o nome pdf_create.

Agora vamos fazer a mesma coisa para a outra Queue, abra a fila pdf_log_queue e vamos criar um binding entre a fila e a exchange pdf_events, ao qual vamos dar o nome pdf_log.

Testando o Recebimento de Mensagens

Vá nas Exchanges e abra a pdf_events, aqui estará mostrando todos os bindings que essa exchange tem.

Se estiver como na imagem acima, podemos dar continuidade e publicar uma mensagem para testarmos, mas como fazemos isso?

Simples, na mesma tela da Exchange, terá um campo chamado “Publish message” (Publicar mensagem), assim vamos poder testar nossas configurações. Vou criar uma mensagem chamada “Hello World!”, e publicar na fila pdf_create.

Se olharmos nesse momento nas filas de mensagens, veremos que a create_pdf_queue, recebeu uma mensagem e que ainda não foi consumida.

Configurando Laravel

Agora vamos realmente fazer nosso projeto enviar e consumir mensagens do RabbitMQ. Existem várias maneiras de fazer o envio e consumo das mensagens, vamos para esse projeto utilizar uma rota para enviar mensagens e outra para consumir.

Primeiro precisamos instalar a biblioteca php-amqplib

composer require php-amqplib/php-amqplib

Vamos criar nossa Controller quer vai gerenciar nossa envio e consumo de mensagens com o RabbitMQ, chamado RabbitMQController, com uma função send e outra consumer, que nesse momento estão vazias.

php artisan make:controller RabbitMQController

namespace App\Http\Controllers;

class RabbitMQController extends Controller
{
    public function send()
    {

    }

    public function consumer()
    {
        
    }
}

Agora em nosso arquivo de rotas route/web.php, vamos criar as rotas de send e consumer.

use App\Http\Controllers\RabbitMQController;

Route::get('/send', [RabbitMQController::class, 'send']);
Route::get('/consumer', [RabbitMQController::class, 'consumer']);

Enviado mensagens para o RabbitMQ

Precisamos criar nossa código para envio das mensagens ao RabbitMQ, lembrando que o arquivo é app/Http/Controllers/RabbitMQController.php.

namespace App\Http\Controllers;

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

class RabbitMQController extends Controller
{
    public function send()
    {
        $connection = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest');

        $message = new AMQPMessage('Hello World Queue');

        $channel = $connection->channel();
        $channel->basic_publish($message, 'pdf_events', 'pdf_create');
        $channel->basic_publish($message, 'pdf_events', 'pdf_log');

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

        echo "Message published to RabbitMQ \n";
    }

    public function consumer()
    {
        
    }
}

Agora em seu Browser acesse http://localhost/send , e você receberá a mensagem de sucesso que colocamos no final da função send “Message published to RabbitMQ”:

E olhando no RabbitMQ nas Queues, veremos que agora temos 2 mensagens na create_pdf_queue e 1 em pdf_log_queue.

Consumindo mensagens do RabbitMQ

Agora vamos adicionar nosso condigo para consumir as mensagens que estão nas filas. Mas nesse código vamos consumir apenas as mensagens que estão na fila pdf_log_queue. Adicionei logs para que possamos ver as mensagens que recebemos e caso apresente alguma falha no processo de consumo das mensagens.

namespace App\Http\Controllers;

use Illuminate\Support\Facades\Log;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Connection\AMQPStreamConnection;

class RabbitMQController extends Controller
{
    public function send()
    {
        $connection = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest');

        $message = new AMQPMessage('Hello World Queue');

        $channel = $connection->channel();
        $channel->basic_publish($message, 'pdf_events', 'pdf_create');
        $channel->basic_publish($message, 'pdf_events', 'pdf_log');

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

        echo "Message published to RabbitMQ \n";
    }

    public function consumer()
    {
        $connection = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest');

        $channel = $connection->channel();

        $callback = function ($msg) {
            Log::info('[x] Received: ', [$msg->getBody()]);
        };

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

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

basic_consume

queue(string): O nome da fila que será consumida.
consumer_tag(string): Identificador que pode ser passado para identificar o consumidor no canal de conexão. Se for passado em branco o servidor criará um identificador automaticamente.
no_local(boolean): Quando definido como true o servidor não envia mensagens para a conexão que a publicou.
no_ack(boolean): Quando definido como true o servidor não espera o reconhecimento do consumidor para a mensagem. Ele entende que a mensagem foi entregue e seu processamento será bem sucedido.
exclusive(boolean): Quando definido como true o consumidor tem acesso exclusivo a esta fila.
nowait(boolean): Quando definido como true o método não vai esperar a resposta do servidor.
callback(callback): Função que será executado para cada mensagem consumida.
arguments(AMQPTable): Passa uma tabela de argumentos opcionais para o consumo.

Vamos testar para ver se nossa aplicação esta consumindo nossas mensagens, vá em seu Browser http://localhost/consumer . Você não receberá nenhuma mensagem no browser, mas você podemos ver nos logs se houve sucesso em storage/logs/laravel.log

Espero que tenham conseguido realizar cada passo com sucesso. E quem quiser verificar o que foi desenvolvido nesse artigo e algumas melhorias veja meu Github.

Artigos - Artigos e dicas feito com PHP + Laravel

Desenvolvedor completamente apaixonado por tecnologia.