SQS & SNS

SQS fornece filas duráveis; SNS fornece pub/sub fan-out. Workers Python usam boto3 para publicar eventos, consultar filas e construir pipelines desacoplados entre serviços e funções Lambda.

Receita

import boto3
 
sqs = boto3.client("sqs")
queue_url = sqs.get_queue_url(QueueName="orders")["QueueUrl"]
sqs.send_message(QueueUrl=queue_url, MessageBody='{"order_id": "1"}')

Quando usar isso:

• Desacoplar produtores e consumidores com backpressure via profundidade da fila
• Notificações fan-out para muitos assinantes com SNS
• Padrões de Retry e DLQ para processamento com falha
• Ponte para Lambda sem infraestrutura de polling customizada

Exemplo de Trabalho

Publicar no SNS, assinar SQS, enviar/receber/excluir com long polling.

import json
import boto3
 
sns = boto3.client("sns")
sqs = boto3.client("sqs")
 
TOPIC_ARN = "arn:aws:sns:us-east-1:123456789012:orders-events"
QUEUE_URL = sqs.get_queue_url(QueueName="orders-worker")["QueueUrl"]
 
def publish_order_event(order_id: str) -> None:
    sns.publish(
        TopicArn=TOPIC_ARN,
        Message=json.dumps({"order_id": order_id, "type": "created"}),
    )
 
def send_to_queue(body: dict) -> None:
    sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=json.dumps(body))
 
def receive_batch(max_messages: int = 5) -> list[dict]:
    resp = sqs.receive_message(
        QueueUrl=QUEUE_URL,
        MaxNumberOfMessages=max_messages,
        WaitTimeSeconds=20,
        MessageAttributeNames=["All"],
    )
    return resp.get("Messages", [])
 
def delete_message(receipt_handle: str) -> None:
    sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=receipt_handle)
 
if __name__ == "__main__":
    send_to_queue({"order_id": "42"})
    for msg in receive_batch():
        print(msg["Body"])
        delete_message(msg["ReceiptHandle"])

O que isso demonstra:

• publish do SNS para fan-out; send_message do SQS para ponto a ponto
• Long polling (WaitTimeSeconds=20) reduz recebimentos vazios e custos
• Excluir apenas após processamento bem-sucedido - caso contrário, a mensagem retorna após o timeout de visibilidade

Mergulho Profundo

Como Funciona

• Filas padrão SQS são "at-least-once" (pelo menos uma vez); filas FIFO adicionam ordenação e desduplicação
• Timeout de visibilidade oculta a mensagem durante o processamento; estenda com change_message_visibility para trabalhos longos
• SNS entrega para assinantes SQS, Lambda, HTTP, e-mail com base na política de assinatura
• DLQ captura mensagens "poison" (problemáticas) após o número máximo de recebimentos

Padrões

Notas Python

# Relatório de falha parcial de lote para fonte de eventos SQS do Lambda (conceitual)
def handler(event, context):
    failures = []
    for record in event["Records"]:
        try:
            process(json.loads(record["body"]))
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

Armadilhas

• Excluir antes de completar o processamento - perda de dados. Correção: excluir após o commit/efeitos colaterais terem sucesso.
• Timeout de visibilidade muito curto - processamento duplicado. Correção: ajustar timeout > p99 de duração do handler.
• Sem DLQ - mensagens "poison" em loop infinito. Correção: política de redrive para DLQ com alarme de profundidade.
• Publicar payloads enormes - limite de 256KB do SQS. Correção: armazenar blob no S3, passar ponteiro na mensagem.
• Polling sem espera longa - consumo de CPU e custos mais altos. Correção: WaitTimeSeconds 10-20.

Alternativas

FAQs

Relacionado

• Lambda - Mapeamento de fonte de eventos SQS
• Noções Básicas de Boto3 - reutilização de cliente em workers
• Retry, Paginação e Limitação - retentativas de API
• S3 - ponteiros para payloads grandes
• Melhores Práticas do SDK da Nuvem - padrões de worker

Versões da Stack: Esta página foi escrita para Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.