Bus de Servicio de Azure: colas, temas, suscripciones y mensajería de IA confiable

radio_button_uncheckedshort_text1. Escenario de mensajería para IA y objetivos

Imagina una API de análisis de documentos cuya inferencia tarda desde unos segundos hasta medio minuto. El tráfico llega en ráfagas, pocos procesadores pueden usar el modelo a la vez y los servicios de notificación, auditoría, métricas y calidad necesitan el resultado. Una cadena sincrónica prolonga la espera, exige disponibilidad simultánea y permite que una entrada lenta o no válida consuma capacidad.

Bus de Servicio de Azure incorpora un agente duradero entre los componentes. La API confirma rápidamente la aceptación, los workers vacían el trabajo a un ritmo controlado y los consumidores posteriores reciben copias independientes. Una entrada problemática permanece disponible para diagnóstico.

• Aplicar nivelación de carga, consumidores competidores, desacoplamiento temporal y publicación/suscripción a IA.
• Elegir colas o temas con suscripciones, sesiones y filtros en el agente.
• Diseñar cuerpos JSON, propiedades, correlación, lotes, expiración y claim check.
• Procesar con Peek-Lock, liquidación explícita, idempotencia, renovación de bloqueo y recuperación de DLQ.
• Probar el flujo en Python con azure-servicebus y Microsoft Entra ID.

Resumen del tema

El agente separa la aceptación rápida del procesamiento variable de IA y ofrece una ruta operativa explícita para entregas, reintentos y errores.

radio_button_uncheckedshort_text2. Espacios de nombres, entidades, protocolos e identidad

Bus de Servicio es un agente de mensajes empresarial completamente administrado. El espacio de nombres es el límite administrativo y de red que contiene colas, temas y suscripciones y expone un punto de conexión como <namespace>.servicebus.windows.net. AMQP 1.0 es el protocolo principal de los SDK modernos y admite liquidación, transacciones, controles de orden y detección de duplicados.

Los clientes pueden usar SAS o Microsoft Entra ID. Para cargas en Azure, prefiere una identidad administrada y el rol de datos más limitado: Azure Service Bus Data Sender, Data Receiver o Data Owner. Así se evitan cadenas de conexión incrustadas; después de migrar todos los clientes, se puede deshabilitar la autenticación local.

Resumen del tema

Trata el espacio de nombres como frontera de seguridad y capacidad, elige las entidades correctas y usa identidad administrada con privilegio mínimo.

radio_button_uncheckedshort_text3. Desacopla la recepción de solicitudes de la inferencia

En una solicitud-respuesta asincrónica, la API valida el sobre, guarda el estado, envía el mensaje y devuelve un identificador de operación. Después, un worker recibe, ejecuta la inferencia, persiste el resultado y publica un evento o permite que el cliente consulte. Productor y consumidor ya no dependen del mismo momento de implementación y escalado.

La arquitectura no elimina la latencia: la vuelve visible y controlable. El producto debe representar los estados aceptado, en ejecución, completado, con error y expirado, proteger la consulta de estado y definir timeout y notificación. La cola no sustituye a la base de datos de resultados.

Resumen del tema

La mensajería asincrónica convierte una inferencia larga en una operación rastreable con estado y resultado persistentes.

radio_button_uncheckedshort_text4. Nivela la carga con una cola duradera

La nivelación absorbe una ráfaga corta en la cola mientras un grupo estable de workers consume al rendimiento sostenible. Protege GPU, memoria y puntos de conexión de modelo limitados y evita mantener cómputo para un pico poco frecuente.

La cola todavía requiere capacidad y supervisión. Un trabajo pendiente creciente aumenta la demora y puede perder valor empresarial. Define capacidad, TTL, alertas y umbrales de escalado con tasas de llegada y proceso, antigüedad aceptable y errores. Haz visible la contrapresión a los llamadores.

Resumen del tema

Usa la cola como búfer controlado entre demanda irregular y capacidad finita, con límites de tamaño y antigüedad.

radio_button_uncheckedshort_text5. Escala con consumidores competidores y desacoplamiento temporal

Varias instancias reciben de la misma cola. Bus de Servicio bloquea cada mensaje para un receptor a la vez y distribuye el trabajo sin un despachador central. Azure Functions, Azure Container Apps o AKS pueden agregar instancias cuando crece el trabajo pendiente.

Si un worker falla antes de liquidar, el bloqueo expira y el mensaje vuelve a estar disponible. Esa recuperación admite duplicados, por lo que los efectos posteriores deben ser idempotentes. El almacenamiento duradero también conserva trabajo durante una implementación o interrupción breve.

Resumen del tema

Los consumidores competidores distribuyen trabajo horizontalmente; el almacenamiento duradero desacopla disponibilidad y la idempotencia hace segura la reentrega.

radio_button_uncheckedshort_text6. Usa la profundidad como contrapresión y señal de escalado

Mensajes activos, antigüedad del más antiguo, tasas de entrada y finalización, latencia, errores y mensajes fallidos describen mejor la salud que la CPU aislada. Crecimiento continuo indica que la llegada supera la finalización; una cola siempre vacía puede significar poca latencia o exceso de capacidad.

Azure Monitor recopila métricas y activa alertas. KEDA en Azure Container Apps o AKS y los desencadenadores de Azure Functions convierten el trabajo pendiente en réplicas. Configura mínimos, máximos, enfriamiento, concurrencia y límites del modelo en conjunto para no trasladar el cuello de botella.

Resumen del tema

Dimensiona con profundidad, antigüedad, rendimiento y límites posteriores; la profundidad es una señal, no el plan completo.

radio_button_uncheckedshort_text7. Elige Standard o Premium y respeta los límites

La cuota incluye cuerpo y propiedades y varía por nivel, protocolo y configuración; consulta la documentación vigente. Documentos, imágenes, audio y artefactos suelen pertenecer a Azure Blob Storage mediante claim check aunque un mensaje grande sea técnicamente posible.

Resumen del tema

Elige el nivel por aislamiento, red, disponibilidad y rendimiento medido; usa las cuotas como límites de validación.

radio_button_uncheckedshort_text8. Elige cola o tema con suscripciones

El receptor no lee directamente del tema: consume una suscripción que actúa como cola virtual y también puede tener consumidores competidores. No uses tres consumidores en una cola cuando los tres deban ver el mensaje; solo uno lo recibirá.

Resumen del tema

Usa cola para una responsabilidad y tema con suscripciones cuando varias responsabilidades necesiten copias independientes.

radio_button_uncheckedshort_text9. Conserva el orden por flujo con sesiones

El orden de llegada no garantiza finalización ordenada con varios workers. Las sesiones agrupan por session_id y conceden un bloqueo exclusivo, ofreciendo FIFO en el grupo y paralelismo entre grupos. Extraer, clasificar y resumir puede compartir el identificador del documento.

La entidad debe crearse con sesiones y cada mensaje debe llevar session_id. Las sesiones reducen concurrencia para claves activas y agregan estado; úsalas solo cuando el orden sea requisito de corrección. Un orquestador puede ser más claro para dependencias complejas.

Resumen del tema

Las sesiones ofrecen proceso exclusivo y ordenado por clave empresarial, manteniendo paralelismo entre claves.

radio_button_uncheckedshort_text10. Distribuye resultados con filtros de suscripción

Cada suscripción comienza con TrueFilter y acepta todo. Sustitúyelo o complétalo con filtros SQL sobre propiedades de sistema y aplicación; los filtros de correlación son eficientes para coincidencias exactas. FalseFilter no acepta mensajes y ayuda cuando todas las reglas serán explícitas.

Propiedades como priority, model_name, document_type, tenant y review_required permiten enrutar sin analizar JSON. Estabiliza nombres y tipos, prueba reglas superpuestas y elimina la regla verdadera predeterminada si anula la selección. El filtro controla entrega, no autorización.

from azure.identity import DefaultAzureCredential
from azure.servicebus.management import (
    ServiceBusAdministrationClient,
    SqlRuleFilter,
)

admin = ServiceBusAdministrationClient(
    "<namespace>.servicebus.windows.net",
    DefaultAzureCredential(),
)

admin.create_rule(
    topic_name="inference-results",
    subscription_name="priority-notifications",
    rule_name="high-priority-only",
    filter=SqlRuleFilter("priority = 'high'"),
)

Resumen del tema

Enruta publicaciones con propiedades estables y reglas probadas; cada suscripción conserva su propio trabajo pendiente, reintentos, escalado y errores.

radio_button_uncheckedshort_text11. Administra remitentes y receptores con el SDK de Python

El paquete azure-servicebus proporciona ServiceBusClient, remitentes de cola/tema y receptores de cola/suscripción. Los administradores de contexto cierran vínculos AMQP de forma predecible. Reutiliza clientes y vínculos duraderos cuando corresponda y sigue la referencia actual porque las API y versiones de Python evolucionan.

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient, ServiceBusMessage

namespace = "<namespace>.servicebus.windows.net"
credential = DefaultAzureCredential()

with ServiceBusClient(namespace, credential) as client:
    with client.get_queue_sender("inference-requests") as sender:
        sender.send_messages(ServiceBusMessage(
            '{"request_id":"req-917","model":"document-analyzer"}',
            content_type="application/json",
            message_id="req-917",
            correlation_id="trace-6d13",
            application_properties={"priority": "high"},
        ))

DefaultAzureCredential funciona con la identidad del desarrollador y con identidad administrada sin cambiar la lógica. Separa los roles Sender y Receiver. Usa get_topic_sender() para temas y get_subscription_receiver(topic_name, subscription_name) para suscripciones.

Resumen del tema

Trata los clientes como recursos administrados, reutiliza conexiones y autentica cada componente con una identidad Microsoft Entra limitada.

radio_button_uncheckedshort_text12. Estructura cuerpos y propiedades de mensajes de IA

El mensaje contiene cuerpo, propiedades de aplicación y propiedades del sistema. JSON es apropiado para request_id, modelo, temperature, max_tokens y referencias de entrada. Define content_type como application/json y versiona el contrato para rechazar o transformar incompatibilidades conscientemente.

Valida esquema, rangos, modelos permitidos, URI y autorización antes de inferencia costosa. No guardes secretos en cuerpo o propiedades; ambos cuentan para la cuota y pueden ser visibles a operadores autorizados.

Resumen del tema

Mantén el contrato versionado en el cuerpo, valores ligeros de enrutamiento en propiedades y la semántica de entrega en propiedades del sistema.

radio_button_uncheckedshort_text13. Correlaciona, rastrea, deduplica y procesa con idempotencia

message_id identifica el mensaje y alimenta la detección de duplicados durante la ventana configurada. correlation_id conecta API, cola, inferencia, publicación y registros. Para OpenTelemetry, propaga traceparent y tracestate en propiedades de aplicación.

La detección protege reenvíos con el mismo message_id, pero no evita una reentrega después de error del receptor o liquidación ambigua. Registra un identificador empresarial en almacenamiento duradero y haz que escrituras, notificaciones y cobros toleren repeticiones. Peek-Lock sigue ofreciendo al menos una entrega; la idempotencia permite un efecto empresarial efectivo una sola vez.

Resumen del tema

Usa message_id contra reenvíos, correlación y trazas para observabilidad e idempotencia duradera contra reentregas.

radio_button_uncheckedshort_text14. Aplica claim check a cargas grandes

Un documento de 500 MB no cabe en un mensaje y las cargas grandes permitidas reducen rendimiento. Carga el documento en Azure Blob Storage privado y publica una referencia opaca con hash, tamaño, tipo, modelo e identificadores. El worker autorizado recupera y comprueba el objeto.

# 1. Upload the large document to private Azure Blob Storage.
blob_uri = upload_with_managed_identity(document_bytes)

# 2. Send only the claim check and routing metadata.
message = ServiceBusMessage(
    json.dumps({
        "request_id": request_id,
        "blob_uri": blob_uri,
        "sha256": payload_hash,
        "model": "document-analyzer",
    }),
    content_type="application/json",
    message_id=request_id,
    correlation_id=correlation_id,
)
sender.send_messages(message)

# 3. The authorized consumer retrieves, validates, and processes the blob.

Prefiere identidad administrada y red privada; si necesitas SAS, limita alcance y duración. Define propiedad, retención, reintento y eliminación para evitar blobs huérfanos o entradas borradas antes de una repetición. Claim check también separa contenido sensible de intermediarios.

Resumen del tema

Guarda entradas grandes o sensibles en almacenamiento protegido y envía una referencia pequeña y verificable con ciclo de vida coordinado.

radio_button_uncheckedshort_text15. Controla vigencia con TTL y rendimiento con lotes

time_to_live expresa cuánto tiempo sigue siendo útil el trabajo. Una recomendación en tiempo real puede expirar pronto; un análisis por lotes, más tarde. Los mensajes expirados pueden pasar a la cola de mensajes fallidos si se habilita esa opción. Los mensajes aplazados tienen comportamiento especial y no deben usarse como almacenamiento permanente.

ServiceBusMessageBatch agrupa mensajes hasta el límite calculado. Si add_message no admite el siguiente, envía el lote, crea otro y vuelve a agregarlo. Un mensaje individual grande todavía requiere claim check. El lote reduce viajes de red, pero no crea una sola transacción empresarial.

from azure.servicebus.exceptions import MessageSizeExceededError

batch = sender.create_message_batch()

for payload in payloads:
    message = ServiceBusMessage(json.dumps(payload))
    try:
        batch.add_message(message)
    except MessageSizeExceededError:
        sender.send_messages(batch)
        batch = sender.create_message_batch()
        batch.add_message(message)

if len(batch) > 0:
    sender.send_messages(batch)

Resumen del tema

Define TTL por utilidad empresarial y agrupa mensajes pequeños hasta el límite del SDK sin confundir lote de transporte y transacción.

radio_button_uncheckedshort_text16. Recibe con Peek-Lock y liquida deliberadamente

from azure.servicebus import ServiceBusReceiveMode

with client.get_queue_receiver(
    queue_name="inference-requests",
    receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
    max_wait_time=30,
) as receiver:
    for message in receiver:
        try:
            payload = json.loads(str(message))
            validate(payload)
            process_idempotently(payload, str(message.message_id))
            receiver.complete_message(message)
        except PermanentPayloadError as error:
            receiver.dead_letter_message(
                message,
                reason="InvalidPayload",
                error_description=str(error),
            )
        except TransientDependencyError:
            receiver.abandon_message(message)

Liquida solo después de los efectos duraderos. Un timeout durante la liquidación es ambiguo: el agente puede haber aplicado la operación sin que el cliente reciba confirmación, otra razón para exigir idempotencia.

Resumen del tema

Peek-Lock prioriza recuperación: completa el éxito, abandona el error transitorio, envía el permanente a DLQ y aplaza solo con plan de recuperación.

radio_button_uncheckedshort_text17. Renueva bloqueos y opera la cola de mensajes fallidos

El bloqueo predeterminado dura un minuto y puede configurarse hasta cinco. Para procesamiento legítimo más largo, renuévalo o usa AutoLockRenewer durante un período limitado. Evita recibir o capturar previamente más mensajes de los que el worker puede terminar. Para tareas siempre largas, registra el trabajo, completa el mensaje pronto y usa una máquina de estados aparte.

from azure.servicebus import AutoLockRenewer

with AutoLockRenewer() as renewer:
    with client.get_queue_receiver("inference-requests") as receiver:
        for message in receiver:
            renewer.register(
                receiver,
                message,
                max_lock_renewal_duration=600,
            )
            run_long_inference(message)
            receiver.complete_message(message)

Cada cola y suscripción tiene una DLQ. Al superar maxDeliveryCount —10 de forma predeterminada— el mensaje llega con MaxDeliveryCountExceeded; la aplicación también puede indicar motivo y descripción. La DLQ retiene mensajes hasta liquidación explícita. Alerta por cantidad y antigüedad, corrige la causa y luego reproduce mediante un proceso idempotente aprobado.

from azure.servicebus import ServiceBusSubQueue

with client.get_queue_receiver(
    queue_name="inference-requests",
    sub_queue=ServiceBusSubQueue.DEAD_LETTER,
    max_wait_time=10,
) as dlq_receiver:
    for message in dlq_receiver:
        inspect(
            message.dead_letter_reason,
            message.dead_letter_error_description,
            message.delivery_count,
            message.correlation_id,
        )
        # Re-submit only after fixing the cause and preserving idempotency.
        replay_if_approved(message)
        dlq_receiver.complete_message(message)

Resumen del tema

Renueva bloqueos para proceso largo acotado, controla intentos y trata la DLQ como flujo observable de reparación.

radio_button_uncheckedshort_text18. Laboratorio, evaluación y lista de producción

El ejercicio original crea un espacio de nombres y una aplicación Flask en Python, usa una cola con Peek-Lock, inspecciona entrada no válida en DLQ y distribuye resultados a suscripciones filtradas. Reserva unos 30 minutos, una suscripción de Azure, Visual Studio Code, Python 3.12 o posterior y la Azure CLI actual.

az group create --name ai200-servicebus-rg --location eastus
az servicebus namespace create   --resource-group ai200-servicebus-rg   --name <globally-unique-namespace>   --location eastus   --sku Standard

az servicebus queue create   --resource-group ai200-servicebus-rg   --namespace-name <namespace>   --name inference-requests   --max-delivery-count 5

az servicebus topic create   --resource-group ai200-servicebus-rg   --namespace-name <namespace>   --name inference-results

python -m venv .venv
python -m pip install --upgrade azure-identity azure-servicebus flask

1. Crea espacio de nombres, cola, tema y suscripciones; configura intentos y filtros explícitamente.
2. Asigna roles Sender o Receiver a identidades administradas y usa DefaultAzureCredential.
3. Envía JSON válido y un mensaje no válido; procesa el trabajo válido de forma idempotente con Peek-Lock.
4. Confirma la DLQ, corrige la causa y reproduce el mensaje una sola vez.
5. Publica un resultado con prioridad y demuestra que solo las suscripciones coincidentes lo reciben.
6. Observa trabajo pendiente, antigüedad, finalizaciones, reintentos, pérdida de bloqueo y DLQ; elimina los recursos al terminar.

Antes de producción, confirma nivel y cuotas, infraestructura como código, privilegio mínimo, autenticación local, red privada, validación del contrato, TTL, ventana de duplicados, sesiones, filtros, almacén de idempotencia, bloqueo, prefetch, reintentos, propietario de DLQ, alertas, runbooks, coste y pruebas de carga en el modelo real.

1. Microsoft Learn: Introducción a la mensajería de Bus de Servicio de Azure
2. Microsoft Learn: Colas, temas y suscripciones
3. Microsoft Learn: Evitar pérdida y procesamiento duplicado
4. Microsoft Learn: Cuotas y límites
5. Microsoft Learn: Colas de mensajes fallidos
6. Microsoft Learn: Biblioteca de Python para Bus de Servicio
7. Azure Architecture Center: Patrón Claim-Check
8. Microsoft Learn: Autenticar aplicaciones con Microsoft Entra ID

Resumen del tema

El diseño completo combina entidad correcta, mensajes pequeños y versionados, identidad, filtros, correlación, Peek-Lock, idempotencia, control del bloqueo, recuperación observable de DLQ y prueba integral.