Barramento de Serviço do Azure: filas, tópicos, assinaturas e mensageria confiável para IA

radio_button_uncheckedshort_text1. Cenário de mensageria para IA e objetivos

Imagine uma API de análise de documentos cuja inferência demora de poucos segundos a meio minuto. O tráfego chega em rajadas, poucos processadores podem usar o modelo ao mesmo tempo e serviços de notificação, auditoria, métricas e qualidade precisam do resultado. Uma cadeia síncrona prolonga a espera, exige disponibilidade simultânea e permite que uma entrada lenta ou inválida ocupe capacidade útil.

O Barramento de Serviço do Azure insere um broker durável entre esses componentes. A API confirma a aceitação rapidamente, os workers drenam o trabalho em um ritmo controlado e os consumidores posteriores recebem cópias independentes. Uma entrada problemática permanece diagnosticável.

• Aplicar nivelamento de carga, consumidores concorrentes, desacoplamento temporal e publicação/assinatura em IA.
• Escolher filas ou tópicos com assinaturas, sessões e filtros no broker.
• Projetar corpos JSON, propriedades, correlação, lotes, expiração e claim check.
• Processar com Peek-Lock, liquidação explícita, idempotência, renovação de bloqueio e recuperação pela DLQ.
• Testar o fluxo em Python com azure-servicebus e Microsoft Entra ID.

Resumo do tópico

O broker separa a aceitação rápida da solicitação do processamento variável de IA e dá um caminho operacional explícito a entregas, repetições e falhas.

radio_button_uncheckedshort_text2. Namespaces, entidades, protocolos e identidade

O Barramento de Serviço é um broker de mensagens empresarial totalmente gerenciado. O namespace é o limite administrativo e de rede que contém filas, tópicos e assinaturas e expõe um endpoint como <namespace>.servicebus.windows.net. AMQP 1.0 é o protocolo principal dos SDKs modernos e viabiliza liquidação, transações, controles de ordem e detecção de duplicidades.

É possível autenticar por SAS ou Microsoft Entra ID. Em cargas no Azure, prefira identidade gerenciada e a função de dados mais restrita: Azure Service Bus Data Sender, Data Receiver ou Data Owner. Isso elimina credenciais embutidas; depois da migração dos clientes, a autenticação local pode ser desativada.

Resumo do tópico

Trate o namespace como limite de segurança e capacidade, escolha as entidades corretas e use identidade gerenciada com privilégio mínimo.

radio_button_uncheckedshort_text3. Desacople a entrada da solicitação da inferência

Em uma solicitação-resposta assíncrona, a API valida o envelope, grava o estado, envia a mensagem e devolve um identificador da operação. Depois, um worker recebe, executa a inferência, persiste o resultado e publica um evento de conclusão ou permite consulta pelo cliente. Produtor e consumidor deixam de depender do mesmo instante de implantação e escala.

A arquitetura não elimina a latência: ela a torna visível e controlável. O produto deve representar os estados aceito, em execução, concluído, falho e expirado, proteger a consulta de estado e definir timeout e notificação. A fila não substitui o banco de resultados.

Resumo do tópico

A mensageria assíncrona transforma uma inferência longa em operação rastreável, com estado e resultado persistidos e evolução independente dos componentes.

radio_button_uncheckedshort_text4. Nivele a carga com uma fila durável

O nivelamento absorve uma rajada curta na fila enquanto um conjunto estável de workers consome na vazão sustentável. Isso protege GPUs, memória e endpoints de modelo limitados e evita manter computação suficiente para um pico raro.

A fila exige capacidade e monitoramento. Um backlog crescente eleva o tempo de conclusão e pode perder valor de negócio. Defina capacidade, TTL, alertas e limites de escala com base nas taxas de chegada e processamento, idade aceitável e comportamento de falha. Torne a contrapressão perceptível aos chamadores.

Resumo do tópico

Use a fila como buffer controlado entre demanda irregular e capacidade finita, com limites explícitos de tamanho e idade do backlog.

radio_button_uncheckedshort_text5. Escale com consumidores concorrentes e desacoplamento temporal

Várias instâncias recebem da mesma fila. O Barramento de Serviço bloqueia cada mensagem para um receptor por vez, distribuindo o trabalho sem um despachante central. Azure Functions, Aplicativos de Contêiner do Azure ou AKS podem adicionar instâncias conforme o backlog cresce.

Se um worker falhar antes da liquidação, o bloqueio expira e a mensagem volta a ficar disponível. Essa recuperação também permite duplicidades, por isso os efeitos posteriores devem ser idempotentes. O armazenamento durável também mantém trabalho durante uma implantação ou interrupção breve.

Resumo do tópico

Consumidores concorrentes distribuem trabalho horizontalmente; armazenamento durável desacopla disponibilidade e idempotência torna a reentrega segura.

radio_button_uncheckedshort_text6. Use profundidade da fila como contrapressão e sinal de escala

Mensagens ativas, idade da mais antiga, taxas de entrada e conclusão, latência, falhas e mensagens mortas descrevem melhor a saúde do que CPU isolada. Crescimento contínuo indica que a chegada supera a conclusão; fila sempre vazia pode significar baixa latência ou excesso de capacidade.

O Azure Monitor coleta métricas e dispara alertas. KEDA nos Aplicativos de Contêiner do Azure ou no AKS e gatilhos do Azure Functions convertem backlog em réplicas. Configure mínimo, máximo, cooldown, concorrência e limites do endpoint de modelo em conjunto para não apenas deslocar o gargalo.

Resumo do tópico

Dimensione com backlog, idade, vazão e limites posteriores em conjunto; profundidade é sinal de pressão, não um plano completo.

radio_button_uncheckedshort_text7. Escolha Standard ou Premium e respeite limites

A cota inclui corpo e propriedades e varia com camada, protocolo e configuração; confira a documentação vigente. Documentos, imagens, áudio e artefatos de modelo normalmente pertencem ao Armazenamento de Blobs do Azure com claim check, mesmo quando uma mensagem grande seria tecnicamente aceita.

Resumo do tópico

Escolha a camada por isolamento, rede, disponibilidade e vazão medida; use as cotas como limites de validação, não como meta de tamanho.

radio_button_uncheckedshort_text8. Escolha fila ou tópico com assinaturas

O receptor não lê o tópico diretamente: lê uma assinatura, que funciona como fila virtual e também pode ter consumidores concorrentes. Não use três consumidores na mesma fila quando os três precisam ver a mensagem; somente um deles a receberá.

Resumo do tópico

Use fila para uma responsabilidade de processamento e tópico com assinaturas quando várias responsabilidades precisarem de cópias independentes.

radio_button_uncheckedshort_text9. Preserve a ordem por fluxo com sessões

A ordem de chegada não garante a conclusão ordenada entre workers. Sessões agrupam mensagens por session_id e concedem um bloqueio exclusivo da sessão, oferecendo FIFO dentro do grupo e paralelismo entre grupos. Extração, classificação e resumo podem compartilhar o identificador do documento.

A entidade deve nascer com sessões habilitadas e toda mensagem precisa de session_id. Sessões reduzem a concorrência de chaves quentes e acrescentam estado e bloqueios; use-as somente quando a ordem for requisito de correção. Dependências complexas talvez sejam mais claras em um orquestrador.

Resumo do tópico

Sessões fornecem processamento exclusivo e ordenado por chave de negócio, mantendo paralelismo entre chaves.

radio_button_uncheckedshort_text10. Distribua resultados com filtros de assinatura

Cada assinatura começa com uma TrueFilter que aceita tudo. Troque-a ou complemente-a com filtros SQL sobre propriedades de sistema e aplicação; filtros de correlação são eficientes para correspondências exatas. FalseFilter não aceita mensagens e ajuda quando todas as regras serão explícitas.

Propriedades como priority, model_name, document_type, tenant e review_required permitem roteamento sem interpretar JSON. Estabilize nomes e tipos, teste regras sobrepostas e remova a regra verdadeira padrão quando ela inviabilizar a seleção. Filtro controla entrega, não autorização.

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

Resumo do tópico

Roteie publicações com propriedades estáveis e regras testadas; cada assinatura mantém backlog, repetição, escala e falha próprios.

radio_button_uncheckedshort_text11. Gerencie remetentes e receptores no SDK do Python

O pacote azure-servicebus oferece ServiceBusClient, remetentes de fila/tópico e receptores de fila/assinatura. Gerenciadores de contexto fecham links AMQP de forma previsível. Reutilize clientes e links duradouros quando adequado e siga a referência atual do SDK, pois APIs e versões de Python evoluem.

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 com a identidade do desenvolvedor e com identidade gerenciada sem alterar a lógica. Separe as funções Sender e Receiver. Use get_topic_sender() para tópicos e get_subscription_receiver(topic_name, subscription_name) para assinaturas.

Resumo do tópico

Trate clientes do SDK como recursos gerenciados, reutilize conexões e autentique cada componente com identidade Microsoft Entra restrita.

radio_button_uncheckedshort_text12. Estruture corpos e propriedades de mensagens de IA

A mensagem contém corpo, propriedades de aplicação e propriedades do sistema. JSON é adequado para request_id, modelo, temperature, max_tokens e referências à entrada ou ao contexto. Defina content_type como application/json e versione o contrato para que incompatibilidades sejam rejeitadas ou transformadas conscientemente.

Valide esquema, faixas, modelos permitidos, URI e autorização antes da inferência cara. Não coloque segredos no corpo ou propriedades; ambos entram na cota e podem ser inspecionados por operadores autorizados.

Resumo do tópico

Mantenha o contrato versionado no corpo, valores leves de roteamento nas propriedades da aplicação e semântica de entrega nas propriedades do sistema.

radio_button_uncheckedshort_text13. Correlacione, rastreie, deduplique e processe com idempotência

message_id identifica a mensagem e alimenta a detecção de duplicidade dentro da janela configurada. correlation_id conecta API, fila, inferência, publicação do resultado e logs. Para OpenTelemetry, propague traceparent e tracestate nas propriedades de aplicação.

A detecção de duplicidade protege reenvios com o mesmo message_id, mas não impede reentrega depois de falha do receptor ou liquidação ambígua. Registre um identificador de negócio em armazenamento durável e torne gravações, notificações e cobranças repetíveis sem efeito extra. Peek-Lock continua sendo entrega pelo menos uma vez; idempotência produz um efeito de negócio efetivamente único quando bem projetada.

Resumo do tópico

Use message_id contra reenvios, correlação e contexto de trace para observabilidade e idempotência durável contra reentregas.

radio_button_uncheckedshort_text14. Aplique claim check a cargas grandes

Um documento de 500 MB não cabe em uma mensagem, e cargas grandes permitidas ainda reduzem a vazão. Envie o documento ao Armazenamento de Blobs do Azure privado e publique apenas uma referência opaca, hash, tamanho, tipo de mídia, modelo e identificadores. O worker autorizado recupera e valida o 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.

Prefira identidade gerenciada e rede privada; se usar SAS, restrinja escopo e duração. Defina propriedade, retenção, repetição e exclusão para não deixar blobs órfãos nem apagar uma entrada antes de um replay. Claim check também evita expor conteúdo sensível a intermediários.

Resumo do tópico

Guarde entradas grandes ou sensíveis em armazenamento de objetos protegido e envie uma referência pequena e verificável com ciclo de vida coordenado.

radio_button_uncheckedshort_text15. Controle validade com TTL e vazão com lotes

time_to_live representa por quanto tempo o trabalho continua útil. Recomendações em tempo real podem expirar cedo; análises em lote, mais tarde. Mensagens expiradas podem ir à fila de mensagens mortas quando a opção está habilitada. Mensagens adiadas têm comportamento especial e não devem virar armazenamento permanente.

ServiceBusMessageBatch agrupa mensagens até o limite calculado. Quando add_message não comporta a próxima, envie o lote, crie outro e tente novamente. Uma mensagem individual grande ainda exige claim check. O lote reduz viagens de rede, mas não cria uma transação de negócio única.

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)

Resumo do tópico

Defina TTL pelo valor de negócio e agrupe mensagens pequenas até o limite do SDK, sem confundir lote de transporte com transação.

radio_button_uncheckedshort_text16. Receba com Peek-Lock e liquide conscientemente

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)

Liquide somente após concluir efeitos duráveis. Timeout durante a liquidação é ambíguo: o broker pode ter aplicado a operação sem a confirmação chegar ao cliente, reforçando a necessidade de idempotência.

Resumo do tópico

Peek-Lock prioriza recuperação: complete o sucesso, abandone falhas transitórias, envie falhas permanentes à DLQ e adie apenas com plano de recuperação.

radio_button_uncheckedshort_text17. Renove bloqueios e opere a fila de mensagens mortas

O bloqueio padrão da entidade dura um minuto e pode chegar a cinco. Para processamento legítimo mais longo, renove-o manualmente ou use AutoLockRenewer por período limitado. Evite receber ou pré-buscar mais mensagens do que o worker concluirá. Para tarefas sempre muito longas, registre o trabalho, conclua a mensagem rapidamente e use uma máquina de estados separada.

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 fila e assinatura tem uma DLQ. Ao ultrapassar maxDeliveryCount — 10 por padrão — a mensagem chega com MaxDeliveryCountExceeded; a aplicação também pode registrar motivo e descrição próprios. A DLQ retém tudo até liquidação explícita. Alerte por quantidade e idade, corrija a causa e só então reproduza por processo idempotente aprovado.

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)

Resumo do tópico

Renove bloqueios para processamento longo limitado, controle tentativas e trate a DLQ como fluxo observável de reparo.

radio_button_uncheckedshort_text18. Laboratório, revisão da avaliação e checklist de produção

O exercício de origem cria namespace e aplicativo Flask em Python, usa fila com Peek-Lock, inspeciona uma entrada inválida na DLQ e distribui resultados a assinaturas filtradas. Reserve cerca de 30 minutos, uma assinatura do Azure, Visual Studio Code, Python 3.12 ou posterior e Azure CLI atual.

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. Crie namespace, fila, tópico e assinaturas; configure tentativas e filtros explicitamente.
2. Atribua funções Sender ou Receiver às identidades gerenciadas e use DefaultAzureCredential.
3. Envie JSON válido e uma mensagem inválida; processe o trabalho válido idempotentemente com Peek-Lock.
4. Confirme a DLQ, corrija a causa e reproduza a mensagem uma única vez.
5. Publique um resultado com prioridade e prove que somente assinaturas compatíveis o recebem.
6. Observe backlog, idade, conclusões, repetições, perda de bloqueio e DLQ; exclua os recursos ao terminar.

Antes da produção, confirme camada e cotas, infraestrutura como código, privilégio mínimo, autenticação local, rede privada, validação de contrato, TTL, janela de duplicidade, sessões, filtros, armazenamento de idempotência, bloqueio, prefetch, retry, responsável pela DLQ, alertas, runbooks, custo e testes de carga no endpoint real.

1. Microsoft Learn: Introdução à mensageria do Barramento de Serviço do Azure
2. Microsoft Learn: Filas, tópicos e assinaturas
3. Microsoft Learn: Evitar perda e processamento duplicado
4. Microsoft Learn: Cotas e limites
5. Microsoft Learn: Filas de mensagens mortas
6. Microsoft Learn: Biblioteca Python do Barramento de Serviço
7. Azure Architecture Center: Padrão Claim Check
8. Microsoft Learn: Autenticar aplicativos com Microsoft Entra ID

Resumo do tópico

O projeto completo combina entidade correta, mensagens pequenas e versionadas, identidade, filtros, correlação, Peek-Lock, idempotência, controle de bloqueio, recuperação observável da DLQ e teste ponta a ponta.