SQS y SNS

SQS proporciona colas duraderas; SNS proporciona pub/sub de tipo fan-out. Los workers de Python utilizan boto3 para publicar eventos, sondear colas y crear pipelines desacoplados entre servicios y funciones Lambda.

Receta

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"}')

Cuándo utilizar esto:

• Desacoplar productores y consumidores con control de flujo (backpressure) mediante la profundidad de la cola.
• Notificaciones de tipo fan-out a muchos suscriptores con SNS.
• Patrones de reintento y DLQ (Dead-Letter Queue) para procesamientos fallidos.
• Puente hacia Lambda sin infraestructura de sondeo personalizada.

Ejemplo de Trabajo

Publicar en SNS, suscribir SQS, enviar/recibir/eliminar con sondeo largo (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"])

Lo que esto demuestra:

• publish de SNS para fan-out; send_message de SQS para punto a punto.
• Sondeo largo (WaitTimeSeconds=20) reduce las recepciones vacías y el coste.
• Eliminar solo después de un procesamiento exitoso; de lo contrario, el mensaje regresa después del tiempo de visibilidad.

Profundización

Cómo Funciona

• Las colas estándar de SQS son "al menos una vez" (at-least-once); las colas FIFO añaden ordenación y deduplicación.
• El tiempo de visibilidad oculta el mensaje durante el procesamiento; extiéndelo con change_message_visibility para trabajos largos.
• SNS entrega a suscriptores de SQS, Lambda, HTTP, correo electrónico basándose en la política de suscripción.
• DLQ captura mensajes "envenenados" (poison messages) después de un número máximo de recepciones.

Patrones

Notas de Python

# Informe de fallo parcial de lote para la fuente de eventos SQS de Lambda (conceptual)
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}

Errores Comunes (Gotchas)

• Eliminar antes de completar el procesamiento - pérdida de datos. Solución: eliminar después de que el commit/efectos secundarios tengan éxito.
• Tiempo de visibilidad demasiado corto - procesamiento duplicado. Solución: ajustar el tiempo de visibilidad a > p99 de la duración del manejador.
• Sin DLQ - los mensajes "envenenados" se repiten indefinidamente. Solución: política de redestinación a DLQ con alarma de profundidad.
• Publicar cargas útiles enormes - límite de 256 KB de SQS. Solución: almacenar el blob en S3, pasar el puntero en el mensaje.
• Sondear sin espera larga - consumo de CPU y mayores costes. Solución: WaitTimeSeconds de 10 a 20.

Alternativas

Preguntas Frecuentes (FAQs)

Relacionado

• Lambda - Mapeo de eventos SQS
• Conceptos básicos de boto3 - Reutilización de clientes en workers
• Reintento, Paginación y Limitación - Reintentos de API
• S3 - Punteros a cargas útiles grandes
• Mejores Prácticas de SDK en la Nube - Patrones de workers

Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+ y uv 0.6+.