OPC UA PubSub sobre MQTT Explicado: Mensagens, Codificações e Topics

O OPC UA PubSub sobre MQTT combina o modelo de informação do OPC UA com a distribuição baseada em broker do MQTT, para que um dispositivo OPC UA publique os dados uma única vez e qualquer número de subscribers os receba por meio de um broker padrão. Neste artigo você vai entender como a Parte 14 da especificação OPC UA estrutura o PubSub, a diferença entre as codificações JSON e UADP, como topics e metadados são organizados e como essas mensagens passam por um broker MQTT comum. Você também vai publicar e decodificar uma NetworkMessage JSON por conta própria.

Por que o OPC UA adicionou o PubSub

O OPC UA clássico é client/servidor: um client abre uma sessão, cria subscriptions e recebe notificações por essa sessão. É poderoso, mas escala mal quando muitos consumidores querem os mesmos dados, e exige que os clients alcancem os servidores diretamente, o que é complicado através de firewalls e até a nuvem, por exemplo entre a rede de automação de uma usina e a TI corporativa. Nosso comparativo OPC UA vs MQTT aborda essa tensão em detalhes.

A Parte 14 da especificação, introduzida com o OPC UA 1.04 e ampliada na 1.05, acrescenta um modelo publish/subscribe. Os publishers enviam mensagens sem saber quem as recebe; os subscribers filtram o que precisam. A Parte 14 define duas famílias de transporte:

• Sem broker, como multicast UDP ou Ethernet (incluindo TSN), para comunicação determinística e de baixa latência entre controladores na mesma rede.
• Com broker, usando MQTT ou AMQP, para distribuir dados a muitos consumidores, entre redes e até a nuvem.

Este artigo trata da segunda família, com o MQTT como transporte.

Os blocos do PubSub

A Parte 14 introduz um vocabulário que vale a pena aprender antes de olhar as mensagens.

A ideia importante é que o payload não é um JSON arbitrário inventado por cada fabricante. É uma estrutura de mensagem definida, com números de sequência, versões de metadados e uma forma de descrever os campos, para que um subscriber genérico consiga interpretar dados de qualquer publisher compatível.

Codificação JSON vs UADP

A Parte 14 define duas codificações de mensagem, e ambas podem ser transportadas sobre MQTT.

Para integração com a nuvem, o JSON é a escolha comum, porque qualquer linguagem consegue interpretá-lo e você pode inspecionar as mensagens diretamente em um explorador de topics. O UADP faz sentido quando a banda importa ou quando as duas pontas são stacks OPC UA.

Anatomia de uma NetworkMessage JSON

Uma mensagem de dados codificada em JSON tem esta aparência (conjunto de campos simplificado para facilitar a leitura):

{
  "MessageId": "8f1c2e7a-0b44-4c55-9a1e-3f0d9d2b6c11",
  "MessageType": "ua-data",
  "PublisherId": "cnc-cell-07",
  "Messages": [
    {
      "DataSetWriterId": 101,
      "SequenceNumber": 4521,
      "MetaDataVersion": { "MajorVersion": 1, "MinorVersion": 3 },
      "Timestamp": "2026-10-16T09:30:12.512Z",
      "MessageType": "ua-keyframe",
      "Payload": {
        "SpindleTemp": 47.3,
        "SpindleSpeed": 12000,
        "PartCount": 1843
      }
    }
  ]
}

Campos-chave para entender:

• MessageType no nível superior distingue dados (ua-data) de metadados (ua-metadata) e, nas versões mais novas, de status e outras mensagens de descoberta.
• PublisherId e DataSetWriterId, juntos, identificam exatamente de qual dataset se trata.
• SequenceNumber permite que os subscribers detectem lacunas e duplicatas.
• MetaDataVersion informa ao subscriber qual versão da descrição dos campos se aplica. Se ela mudar, o subscriber deve buscar os novos metadados antes de confiar nas posições ou nos tipos dos campos.
• Key frames vs delta frames: um key frame carrega todos os campos; um delta frame carrega apenas os campos que mudaram, o que reduz o tráfego de dados que mudam devagar.

Os valores do payload podem ser codificados em formas mais compactas ou mais verbosas (a especificação distingue codificações reversíveis e não reversíveis, e a 1.05 refina isso). A forma compacta traz valores simples, como acima; a forma verbosa envolve cada valor com seu tipo, status e timestamps, para que ele possa ser decodificado de volta em um Variant OPC UA exato.

Topics e metadados no broker

No OPC UA 1.04, o topic MQTT (chamado de queue name) é simplesmente configurado em cada WriterGroup ou DataSetWriter. O OPC UA 1.05 acrescenta uma árvore de topics padrão recomendada, para que os subscribers possam descobrir os publishers de forma previsível. Ela segue este formato:

<prefix>/<encoding>/<message type>/<PublisherId>/<WriterGroup>/<DataSetWriter>

opcua/json/data/cnc-cell-07/fast/spindle
opcua/json/metadata/cnc-cell-07/fast/spindle
opcua/json/status/cnc-cell-07

O prefixo padrão é opcua, mas é configurável, o que importa quando você compartilha um broker com outros sistemas. Vale a pena seguir três convenções:

1. As mensagens de metadados são publicadas como retidas, para que um subscriber que se conecte depois receba a descrição dos campos imediatamente.
2. As mensagens de status são retidas e combinadas com um Last Will, para que os subscribers saibam se um publisher está online.
3. O QoS é configurável por writer group, mapeado para QoS 0, 1 ou 2 do MQTT (melhor esforço, pelo menos uma vez, exatamente uma vez).

Nada disso exige suporte especial do broker. Retained messages, Last Will, wildcards e QoS são recursos padrão do MQTT, e é exatamente por isso que a Parte 14 escolheu o MQTT como transporte. Para relembrar as garantias de entrega, veja os níveis de QoS do MQTT explicados.

Como as mensagens passam por um broker padrão

Do ponto de vista do broker, um publisher OPC UA PubSub é apenas mais um client MQTT. Ele se conecta, opcionalmente se autentica, define um Last Will e publica payloads em topics. O broker não interpreta a estrutura OPC UA; ele roteia bytes com base em topics e assinaturas. Isso significa que qualquer broker MQTT compatível com o padrão pode transportar tráfego OPC UA PubSub, incluindo um broker MQTT gerenciado como o ConnectMQ.

Um subscriber normalmente:

1. Assina o topic de metadados, recebe os metadados retidos e monta um mapa de campos.
2. Assina o topic de dados ou um wildcard como opcua/json/data/#.
3. Compara o MetaDataVersion com os metadados em cache e verifica o SequenceNumber em busca de lacunas.
4. Aplica os delta frames ao último key frame conhecido para manter o estado atual.

Na prática: publique e decodifique uma NetworkMessage JSON

O servidor MQTT público (gratuito), que serve apenas para testes e protótipos, limita os topics a oito níveis, por isso o exemplo substitui o prefixo opcua por um prefixo exclusivo próprio. Instale primeiro o paho-mqtt 2.x com pip install "paho-mqtt>=2.0".

O publisher envia uma mensagem de metadados retida e depois um key frame a cada dois segundos:

# pubsub_publisher.py - emite mensagens JSON no estilo da Parte 14 do OPC UA via MQTT
import json, random, time, uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/c3d9/json"          # substitui o padrão "opcua/json"
PUBLISHER_ID = "cnc-cell-07"
DATA_TOPIC = f"{PREFIX}/data/{PUBLISHER_ID}/fast/spindle"
META_TOPIC = f"{PREFIX}/metadata/{PUBLISHER_ID}/fast/spindle"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"uapub-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

metadata = {
    "MessageId": str(uuid.uuid4()), "MessageType": "ua-metadata",
    "PublisherId": PUBLISHER_ID, "DataSetWriterId": 101,
    "MetaData": {
        "Name": "Spindle",
        "Fields": [
            {"Name": "SpindleTemp", "BuiltInType": 11},   # Double
            {"Name": "SpindleSpeed", "BuiltInType": 11},
            {"Name": "PartCount", "BuiltInType": 7}       # UInt32
        ],
        "ConfigurationVersion": {"MajorVersion": 1, "MinorVersion": 3}
    }
}
client.publish(META_TOPIC, json.dumps(metadata), qos=1, retain=True)

seq, parts = 0, 1840
while True:
    seq += 1
    parts += random.choice([0, 0, 1])
    msg = {
        "MessageId": str(uuid.uuid4()), "MessageType": "ua-data", "PublisherId": PUBLISHER_ID,
        "Messages": [{
            "DataSetWriterId": 101, "SequenceNumber": seq,
            "MetaDataVersion": {"MajorVersion": 1, "MinorVersion": 3},
            "Timestamp": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "MessageType": "ua-keyframe",
            "Payload": {"SpindleTemp": round(random.uniform(44, 50), 2),
                        "SpindleSpeed": random.choice([0.0, 8000.0, 12000.0]),
                        "PartCount": parts}
        }]
    }
    client.publish(DATA_TOPIC, json.dumps(msg), qos=1)
    time.sleep(2)

O subscriber armazena os metadados em cache, verifica as versões e detecta lacunas na sequência:

# pubsub_subscriber.py - decodifica NetworkMessages JSON do OPC UA a partir do MQTT
import json, uuid
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/c3d9/json"
meta_versions, last_seq = {}, {}

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/metadata/#", qos=1)
    client.subscribe(f"{PREFIX}/data/#", qos=1)

def on_message(client, userdata, message):
    nm = json.loads(message.payload)
    if nm["MessageType"] == "ua-metadata":
        key = (nm["PublisherId"], nm["DataSetWriterId"])
        meta_versions[key] = nm["MetaData"]["ConfigurationVersion"]
        print("metadata", key, [f["Name"] for f in nm["MetaData"]["Fields"]])
        return
    for dsm in nm.get("Messages", []):
        key = (nm["PublisherId"], dsm["DataSetWriterId"])
        if meta_versions.get(key) != dsm["MetaDataVersion"]:
            print("warning: metadata missing or outdated for", key)
        prev = last_seq.get(key)
        if prev is not None and dsm["SequenceNumber"] != prev + 1:
            print(f"gap detected: {prev} -> {dsm['SequenceNumber']}")
        last_seq[key] = dsm["SequenceNumber"]
        print(dsm["Timestamp"], dsm["Payload"])

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"uasub-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

Execute o publisher e inicie o subscriber a qualquer momento. Como os metadados são retidos, o subscriber imprime primeiro a lista de campos e depois os dados. Substitua c3d9 por uma string aleatória sua.

Considerações práticas

• Convivência com um Unified Namespace. Os topics do OPC UA PubSub são organizados por publisher e writer group, não pela hierarquia da planta. Muitas equipes mantêm a árvore PubSub como está e rodam um pequeno serviço que republica campos selecionados em um namespace no estilo ISA-95, que também pode alimentar um gêmeo digital da planta; veja nosso guia sobre topics MQTT ISA-95.
• Tamanho das mensagens. Key frames com muitos campos podem ficar grandes. Verifique o limite de tamanho de mensagem do broker no seu plano e prefira delta frames ou vários writer groups para datasets extensos.
• Segurança. Sobre MQTT, você protege o transporte com TLS e credenciais do broker. A Parte 14 também define assinatura e criptografia no nível da mensagem com um serviço de chaves, usadas principalmente com UADP.
• Ferramentas. Stacks open source como o open62541 incluem PubSub com suporte a MQTT, e vários CLPs e gateways comerciais conseguem publicar JSON da Parte 14 diretamente.

Para o contexto mais amplo de onde isso se encaixa na arquitetura de uma planta, leia nosso guia de IoT industrial com MQTT.

Comece com o ConnectMQ

Aponte seu publisher OPC UA PubSub para um broker que você não precisa operar. Crie um gratuitamente em app.connectmq.com no plano Serverless (Free), sem cartão de crédito, use o usuário, a senha e o client ID gerados com TLS na porta 8883 e inspecione suas NetworkMessages no explorador de topics ao vivo do portal. O ConnectMQ suporta MQTT 3.1.1 e 5.0, retained messages e Last Will, tudo aquilo de que a Parte 14 depende.

Perguntas frequentes

O broker MQTT precisa entender OPC UA para o PubSub funcionar?

Não. O broker roteia as mensagens por topic e trata os payloads como bytes. Qualquer broker MQTT compatível com o padrão pode transportar tráfego OPC UA PubSub.

Devo usar codificação JSON ou UADP no OPC UA PubSub sobre MQTT?

Use JSON quando sistemas de TI ou de nuvem consomem os dados, porque é legível e fácil de interpretar. Use UADP quando a banda for limitada ou as duas pontas forem stacks OPC UA.

Qual é a diferença entre um key frame e um delta frame?

Um key frame contém todos os campos do dataset; um delta frame contém apenas os campos que mudaram. Os subscribers aplicam os delta frames ao último key frame para manter o estado completo.

Por que as mensagens de metadados devem ser retidas?

Para que um subscriber que se conecte depois de o publisher ter começado ainda receba imediatamente a descrição dos campos, sem esperar que o publisher a reenvie.