O Que É uma Fila de Mensagens e Como Usar

Um endpoint de checkout que redimensiona a imagem de um produto, envia um email de confirmação e atualiza um mecanismo de recomendação antes de responder com 200 OK é tão rápido quanto sua etapa mais lenta. Se uma delas der timeout, a requisição inteira falha. Uma fila de mensagens permite que o endpoint entregue esse trabalho como mensagens e responda na hora, enquanto workers separados as processam no próprio ritmo.

O que uma fila de mensagens realmente faz

Uma fila de mensagens fica entre duas partes de um sistema que não precisam rodar no mesmo momento. Um producer publica uma mensagem descrevendo um trabalho a ser feito. A fila a guarda. Um consumer a pega, processa e confirma. Os dois nunca conversam diretamente.

Pense numa caixa de correio. Quem envia deposita a carta e vai embora; quem lê confere quando está livre, e se fica fora por uma hora, as cartas esperam.

Disso vêm três coisas que uma chamada de função direta não consegue dar:

• Desacoplamento. O endpoint de checkout não precisa saber como funciona o redimensionamento de imagem, nem que é lento. Ele publica {"event": "order_placed", "order_id": 4821} e segue em frente.
• Buffering. Se 500 pedidos chegam no mesmo segundo, a fila absorve o pico. Os workers os processam no ritmo que conseguem sustentar, em vez de 500 requisições bloqueadas de uma vez.
• Escalonamento e falhas independentes. Rode três workers de redimensionamento e um de email, reinicie um sem mexer no outro, faça deploy do serviço de checkout sem redistribuir seus workers.

Rodando o RabbitMQ no Docker

O RabbitMQ é o message broker de propósito geral mais usado e uma escolha razoável se você ainda não tem uma fila. Baixe a imagem management para ter também uma UI web junto com o broker:

1
2
3
4
5

docker run -d --hostname mq --name rabbitmq \
  -e RABBITMQ_DEFAULT_USER=app \
  -e RABBITMQ_DEFAULT_PASS=changeme \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:4-management

A porta 5672 é o protocolo AMQP ao qual sua aplicação se conecta. A porta 15672 é a UI de gerenciamento: abra http://localhost:15672, entre com app / changeme, e você acompanha filas, taxa de mensagens e conexões ao vivo. O login padrão guest/guest só funciona de dentro do container, então defina seu próprio usuário para qualquer coisa que você acessar a partir do host.

Se você nunca usou o Docker, O Que é Docker e Para Que Serve explica imagens, containers e portas. Para um stack real, rode o broker junto com sua aplicação usando o Compose — veja O Que É Docker Compose e Como Usar:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

services:
  rabbitmq:
    image: rabbitmq:4-management
    environment:
      - RABBITMQ_DEFAULT_USER=app
      - RABBITMQ_DEFAULT_PASS=changeme
    ports:
      - "5672:5672"
      - "15672:15672"

  worker:
    build: ./worker
    depends_on:
      - rabbitmq
    environment:
      - RABBITMQ_URL=amqp://app:changeme@rabbitmq:5672/

O worker alcança o broker pelo hostname rabbitmq. O Compose coloca os dois serviços na mesma rede user-defined, então o nome do serviço também funciona como hostname resolvível — o mecanismo descrito em O Que É uma Rede Docker e Como Usar.

Publicando e consumindo uma mensagem

pika é o cliente Python padrão para o protocolo do RabbitMQ (AMQP 0-9-1). Instale com pip install pika. Um producer que publica uma tarefa fica assim:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host="localhost",
        credentials=pika.PlainCredentials("app", "changeme"),
    )
)
channel = connection.channel()

channel.queue_declare(queue="resize_image", durable=True)

channel.basic_publish(
    exchange="",
    routing_key="resize_image",
    body='{"order_id": 4821, "image_url": "https://example.com/p/4821.jpg"}',
    properties=pika.BasicProperties(delivery_mode=pika.DeliveryMode.Persistent),
)
connection.close()

durable=True em queue_declare diz ao RabbitMQ para manter a fila em si depois de um restart do broker. delivery_mode=pika.DeliveryMode.Persistent diz para gravar cada mensagem em disco em vez de mantê-la só em memória. Sem os dois, um docker restart no broker apaga silenciosamente tudo que ainda não tinha sido consumido.

O consumer pega mensagens da mesma fila e só confirma cada uma quando o trabalho está realmente feito:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host="localhost",
        credentials=pika.PlainCredentials("app", "changeme"),
    )
)
channel = connection.channel()
channel.queue_declare(queue="resize_image", durable=True)
channel.basic_qos(prefetch_count=1)

def handle_message(ch, method, properties, body):
    resize_image(body)  # o trabalho de verdade
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue="resize_image", on_message_callback=handle_message)
channel.start_consuming()

Duas configurações fazem o trabalho de confiabilidade aqui:

• basic_qos(prefetch_count=1) impede que o RabbitMQ entregue uma segunda mensagem a um worker antes que ele confirme a primeira. Sem isso, um worker ocupado pode ficar com dez mensagens enquanto um livre não recebe nenhuma.
• basic_ack depois do trabalho, não antes. Se o processo cai no meio do redimensionamento, a mensagem nunca foi confirmada, então o RabbitMQ a reentrega a outro worker em vez de perdê-la. Deixe auto_ack desativado, que já é o padrão do pika.

Rode três cópias do script consumer e o RabbitMQ divide a fila entre elas em round-robin. Essa é toda a história de escalonamento desse padrão: nenhum código de coordenação, nenhum estado compartilhado entre workers.

Algumas mensagens falham toda vez que são entregues: um payload malformado, uma chamada a um serviço que não existe mais. A reentrega transforma isso num loop infinito pelos seus consumers. Configure x-dead-letter-exchange na fila, e as mensagens que você rejeita sem recolocar na fila (basic_nack com requeue=False) vão para um exchange separado, onde você as inspeciona manualmente.

Como os exchanges do RabbitMQ roteiam mensagens

Os exemplos acima publicam com exchange="", o default exchange, que roteia uma mensagem direto para a fila indicada em routing_key. Isso cobre a maioria dos casos de fila de tarefas. O modelo real do RabbitMQ coloca um exchange na frente de cada fila, e trocar o tipo de exchange muda o comportamento de roteamento:

Um exchange fanout é como você consegue publish/subscribe com o RabbitMQ: você publica order_placed uma vez só, e o serviço de email, o de analytics e o de detecção de fraude recebem cada um sua própria cópia pela própria fila, em vez de disputar as mesmas mensagens.

Quando uma fila de mensagens é a ferramenta errada

• Quem chama precisa da resposta para responder também. Uma fila serve para “fazer isso eventualmente”, não para “calcular isso agora”. Se seu endpoint de checkout precisa do custo de frete calculado antes de responder, isso é uma chamada síncrona, não um job em fila.
• Você precisa de ordem estrita em toda a fila. O RabbitMQ garante ordem por fila com um único consumer, mas adicione um segundo consumer para throughput e duas mensagens podem terminar fora de ordem. Se a sequência importa, como num event log ou numa máquina de estados, você precisa de uma ferramenta feita para streams ordenados, como partições do Kafka com chave pelo ID da entidade.
• O job é pequeno, síncrono, e você ainda não roda um broker. Uma thread em background ou um job runner in-process representam menos superfície operacional do que subir e monitorar o RabbitMQ para uma tarefa do tamanho de um cron.
• Você já tem Redis e tolera perder um job de vez em quando. Redis Streams ou uma biblioteca como BullMQ te dão uma fila sem infraestrutura nova, ao custo de garantias de entrega mais fracas se o broker cair. Veja primeiro o que o Redis já faz no seu stack: Como Funciona o Cache Redis e Como Usá-lo.

Movendo sua primeira tarefa para uma fila

Escolha algo que bloqueia uma requisição sem necessidade: o email de confirmação, a miniatura, o ping de analytics. Rode o RabbitMQ no Docker, declare uma fila durable para essa tarefa, mova-a. Configure prefetch_count=1 e confirme manualmente desde o primeiro dia, porque adicionar confiabilidade depois que uma queda já engoliu uma mensagem custa muito mais do que escrever essas duas linhas agora. Só recorra a um exchange fanout ou topic quando tiver um segundo consumer que quer o mesmo evento.

Artigos relacionados

• Como Funciona o Cache Redis e Como Usá-lo
• O Que É Docker Compose e Como Usar
• O Que É uma Rede Docker e Como Usar