Blog - Felipe Miiller

Introdução

O RabbitMQ surge como uma solução robusta e amplamente adotada para gerenciar essa comunicação assíncrona, atuando como um "message broker" que facilita a troca de mensagens entre aplicações distribuídas. Ele garante que as mensagens sejam entregues de forma confiável, mesmo em cenários de alta demanda ou falhas de serviço.

Mostraremos os conceitos centrais do RabbitMQ: Exchanges e Queues . Exploraremos suas funcionalidades, como interagem entre si e, mais importante, como você pode configurá-los e utilizá-los em suas aplicações NestJS, aproveitando a poderosa biblioteca @golevelup/nestjs-rabbitmq .

O que é RabbitMQ?

RabbitMQ é um software de código aberto que implementa o Advanced Message Queuing Protocol (AMQP), um protocolo de mensagens que permite que aplicações se comuniquem de forma assíncrona. Ele atua como um intermediário, recebendo mensagens de "produtores" e entregando-as a "consumidores". Essa abordagem desacopla os serviços, tornando-os mais independentes e flexíveis. Em vez de um serviço chamar diretamente outro, ele envia uma mensagem para o RabbitMQ, que se encarrega de roteá-la para o destino correto. Isso é crucial para sistemas distribuídos, onde a falha de um serviço não deve impactar a operação de outros.

Exchanges: O Coração do Roteamento de Mensagens

No RabbitMQ, uma Exchange é a primeira parada de uma mensagem enviada por um produtor. Ela não armazena mensagens, mas sim as recebe e as roteia para uma ou mais Queues, com base em regras predefinidas. A forma como uma Exchange roteia as mensagens é determinada pelo seu tipo. Compreender os diferentes tipos de Exchanges é fundamental para projetar um sistema de mensagens eficiente.

Tipos de Exchanges:

Direct Exchange: Roteia mensagens para Queues cuja "binding key" (chave de ligação) corresponde exatamente à "routing key" (chave de roteamento) da mensagem. É ideal para comunicação ponto a ponto ou para rotear mensagens para uma Queue específica.
Fanout Exchange: Roteia mensagens para todas as Queues que estão ligadas a ela, independentemente da routing key. É útil para cenários de "publish/subscribe" (publicar/assinar), onde uma mensagem precisa ser entregue a múltiplos consumidores.
Topic Exchange: Roteia mensagens para Queues com base em padrões de correspondência entre a routing key da mensagem e a binding key da Queue. A binding key pode conter curingas ( para uma palavra e # para zero ou mais palavras), tornando-a flexível para roteamento complexo. É comumente usada para roteamento baseado em tópicos hierárquicos.
Headers Exchange: Roteia mensagens com base nos cabeçalhos da mensagem, em vez da routing key. Permite um roteamento mais complexo e flexível, onde múltiplos atributos da mensagem podem ser usados para determinar o destino.

Queues: Onde as Mensagens Esperam

Uma Queue (fila) é um buffer onde as mensagens são armazenadas até que um consumidor esteja pronto para processá-las. As Queues são os destinos finais das mensagens roteadas pelas Exchanges. Elas garantem a persistência das mensagens (se configurado) e a entrega confiável, mesmo que os consumidores estejam offline ou sobrecarregados.

Características das Queues:

Durabilidade: Queues podem ser declaradas como duráveis, o que significa que elas sobreviverão a reinícios do broker RabbitMQ. Mensagens persistentes publicadas em Queues duráveis também sobreviverão a reinícios.
Exclusividade: Uma Queue exclusiva só pode ser acessada pela conexão que a declarou e é excluída quando a conexão é fechada.
Auto-delete: Uma Queue com auto-delete é excluída automaticamente quando o último consumidor se desconecta.
Prefetch Count: Define o número máximo de mensagens que um consumidor pode receber de uma Queue de uma vez. Isso ajuda a controlar o fluxo de mensagens e a evitar que um consumidor fique sobrecarregado.

Configurando RabbitMQ com NestJS e `@golevelup/nestjs-rabbitmq`

A biblioteca @golevelup/nestjs-rabbitmq simplifica a integração do RabbitMQ em aplicações NestJS, fornecendo decoradores e serviços que abstraem grande parte da complexidade. Vamos explorar como configurar Exchanges e Queues usando esta biblioteca. Essa biblioteca é bem mais robusta que a fornecida pelo Nest.

Instalação

Primeiro, instale o pacote:

npm install --save @golevelup/nestjs-rabbitmq

Configuração do Módulo

Para configurar o RabbitMQModule, você o importa no seu módulo principal (ou em um módulo compartilhado) e usa o método forRoot ou forRootAsync:

import { Module } from '@nestjs/common';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRoot({
      exchanges: [
        {
          name: 'my-exchange',
          type: 'topic',
        },
        {
          name: 'fanout-exchange',
          type: 'fanout',
        },
      ],
      uri: 'amqp://rabbitmq:rabbitmq@localhost:5672', // Sua string de conexão RabbitMQ
      connectionInitOptions: { wait: false }, // Não espera a conexão inicial para iniciar a aplicação
      channels: {
        'channel-1': {
          prefetchCount: 15,
          default: true,
        },
        'channel-2': {
          prefetchCount: 2,
        },
      },
    }),
  ],
  controllers: [],
  providers: [],
})
export class AppModule {}

Neste exemplo:

exchanges: Um array onde você declara as Exchanges que serão usadas. A biblioteca as criará automaticamente se não existirem. Você define o name e o type (e.g., topic, fanout, direct, headers).
uri: A string de conexão para o seu servidor RabbitMQ.
connectionInitOptions: Opções para o gerenciamento da conexão. wait: false permite que a aplicação inicie mesmo que o RabbitMQ não esteja disponível imediatamente, tentando reconectar em segundo plano.
channels: Permite configurar canais AMQP com diferentes prefetchCount (número de mensagens que um consumidor pode receber de uma vez). Isso é útil para otimizar o consumo de mensagens.

Publicando Mensagens

Para publicar mensagens, você pode injetar o AmqpConnection no seu serviço ou controller:

import { Injectable } from '@nestjs/common';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';

@Injectable()
export class MessageProducerService {
  constructor(private readonly amqpConnection: AmqpConnection) {}

  async publishMessage() {
    // Publica uma mensagem na 'my-exchange' com a routing key 'my.routing.key'
    await this.amqpConnection.publish(
      'my-exchange',
      'my.routing.key',
      { message: 'Hello RabbitMQ!' },
    );
    console.log('Mensagem publicada!');
  }

  async publishFanoutMessage() {
    // Publica uma mensagem na 'fanout-exchange' (não precisa de routing key, pois fanout envia para todas as queues ligadas)
    await this.amqpConnection.publish(
      'fanout-exchange',
      '', // Routing key vazia para fanout
      { event: 'UserCreated', userId: 123 },
    );
    console.log('Mensagem fanout publicada!');
  }
}

Consumindo Mensagens

Para consumir mensagens, você usa o decorador @RabbitSubscribe em um método do seu serviço ou controller. A biblioteca criará automaticamente a Queue e a ligará à Exchange especificada.

import { Injectable } from '@nestjs/common';
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';

@Injectable()
export class MessageConsumerService {
  @RabbitSubscribe({
    exchange: 'my-exchange',
    routingKey: 'my.routing.key', // Ou um padrão como 'my.*' para topic exchange
    queue: 'my-queue', // Nome da fila que será criada/consumida
  })
  public async handleMyMessage(msg: {}) {
    console.log(`Mensagem recebida de 'my-exchange': ${JSON.stringify(msg)}`);
    // Lógica de processamento da mensagem
  }

  @RabbitSubscribe({
    exchange: 'fanout-exchange',
    routingKey: '', // Routing key vazia para fanout
    queue: 'fanout-consumer-queue', // Cada consumidor fanout deve ter sua própria fila
  })
  public async handleFanoutMessage(msg: {}) {
    console.log(`Mensagem fanout recebida: ${JSON.stringify(msg)}`);
    // Lógica de processamento do evento
  }
}