Arquitetura Lógica do AMQP

O modelo do protocolo AMQP (Advanced Message Queuing Protocol) é organizado em camadas hierárquicas: Broker → Virtual Host → Exchange → Queue. Essa estrutura assemelha-se a uma divisão geográfica como país, estado e cidade, permitindo isolar ambientes, separar aplicações e gerenciar fluxos de mensagens complexos com alta escalabilidade e baixo acoplamento.

Conceitos Fundamentais do AMQP

• Mensagem (Message): Unidade básica de dados transmitida. Composta por um corpo (payload binário) e um cabeçalho com atributos como routing key, prioridade, modo de entrega (persistente ou volátil), entre outros. O conteúdo do corpo é opaco ao broker.
• Publicador (Publisher): Aplicação cliente que envia mensagens ao broker através de um exchange.
• Consumidor (Consumer): Aplicação que recebe mensagens de uma fila específica, processando-as conforme necessário.
• Exchange: Componente intermediário que recebe mensagens do publicador e as roteia para filas com base em regras definidas. Os tipos mais comuns são:
  • direct: Roteamento exato pela chave de rota.
  • topic: Roteamento por padrão de tópico (wildcard).
  • fanout: Transmissão para todas as filas vinculadas (broadcast).
• Binding: Ligação entre uma fila e um exchange, definida por uma chave de ligação (binding key). O exchange utiliza essa chave, comparando-a com a routing key da mensagem, para decidir o destino.
• Fila (Queue): Armazena mensagens até que sejam consumidas. Pode ser durável ou volátil, e suporta múltiplos consumidores. Uma mensagem pode ser entregue a mais de uma fila, dependendo das ligações.
• Routing Key: Identificador anexado à mensagem que orienta o exchange sobre onde encaminhá-la. Se não houver correspondência com nenhum binding, a mensagem pode ser descartada ou retornada.
• Conexão (Connection): Canal TCP de longa duração entre o cliente e o servidor RabbitMQ. É cara para criar, logo deve ser reutilizada.
• Canal (Channel): Fluxo virtual dentro de uma conexão. Permite multiplexar operações (publicar, consumir, declarar) sem abrir novas conexões TCP. Todas as ações do AMQP ocorrem dentro de um canal.
• Host Virtual (vHost): Espaço isolado com suas próprias filas, exchanges, bindings e permissões. Funciona como um "servidor lógico" dentro de um mesmo broker físico. O vHost padrão é /.
• Broker: O servidor de mensagens propriamente dito — entidade responsável por receber, armazenar e encaminhar mensagens.

Fluxo de Envio e Consumo de Mensagens

1. O publicador envia uma mensagem ao exchange, informando uma routing key.
2. O exchange verifica sua tabela de bindings e identifica quais filas estão associadas com chaves compatíveis.
3. A mensagem é coolcada na(s) fila(s) correspondente(s).
4. O consumidor conecta-se à fila e retira mensagens para processamento.

Exemplo Prático em PHP

A seguir, implementações simplificadas de publicador, consumidor e configuração comum usando a biblioteca AMQP do PHP.

Configuração Compartilhada: config.php

<?php
class RabbitMQManager {
    private $connectionParams = [
        'host'     => '127.0.0.1',
        'port'     => 5672,
        'login'    => 'guest',
        'password' => 'guest',
        'vhost'    => '/'
    ];

    public $exchangeName = 'event_exchange';
    public $routingKey   = 'log.route';

    public function getConnection() {
        $connection = new AMQPConnection($this->connectionParams);
        if (!$connection->connect()) {
            throw new Exception("Falha ao conectar ao broker.");
        }
        return $connection;
    }

    public function createChannel($conn) {
        return new AMQPChannel($conn);
    }

    public function setupExchange($channel) {
        $exchange = new AMQPExchange($channel);
        $exchange->setName($this->exchangeName);
        $exchange->setType(AMQP_EX_TYPE_DIRECT);
        $exchange->setFlags(AMQP_DURABLE);
        $exchange->declareExchange();
        return $exchange;
    }
}

Produtor: publisher.php

<?php
require_once 'config.php';

$manager = new RabbitMQManager();
$conn    = $manager->getConnection();
$channel = $manager->createChannel($conn);
$exchange = $manager->setupExchange($channel);

// Iniciar transação para garantir entrega
$channel->startTransaction();

for ($id = 1; $id <= 5; $id++) {
    sleep(1);
    $payload = json_encode([
        'event_id'    => $id,
        'type'        => 'notification.sms',
        'status'      => 'sent',
        'timestamp'   => date('c'),
        'retry_count' => 0
    ]);

    $result = $exchange->publish(
        $payload,
        $manager->routingKey,
        AMQP_NOPARAM,
        ['delivery_mode' => 2] // Persistente
    );

    echo "Mensagem {$id} enviada: " . ($result ? 'OK' : 'FALHA') . "\n";
}

$channel->commitTransaction(); // Confirma envios
$conn->disconnect();

Consumidor: consumer.php

<?php
require_once 'config.php';

$manager = new RabbitMQManager();
$conn    = $manager->getConnection();
$channel = $manager->createChannel($conn);
$exchange = $manager->setupExchange($channel);

$queue = new AMQPQueue($channel);
$queue->setName('sms_queue');
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();

// Vincular fila ao exchange com a chave de roteamento
$queue->bind($exchange->getName(), $manager->routingKey);

echo "Aguardando mensagens...\n";

$queue->consume(function ($messageEnvelope) {
    $data = $messageEnvelope->getBody();
    file_put_contents('logs/sms.log', $data . "\n", FILE_APPEND | LOCK_EX);
}, AMQP_AUTOACK); // Autoconfirmação habilitada

// Nunca alcança esta linha (loop infinito)
$conn->disconnect();

Este exemplo demonstra a criação de um ambiente funcional de mensageria com troca direta, filas duráveis e entrega confiável. A separação clara entre componentes permite fácil manutenção, teste e expansão para cenários mais complexos como balanceamento de carga e failover.