OPC UA PubSub über MQTT erklärt: Nachrichten, Kodierungen und Topics

OPC UA PubSub über MQTT verbindet das Informationsmodell von OPC UA mit der broker-basierten Verteilung von MQTT, sodass ein OPC-UA-Gerät Daten einmal veröffentlichen kann und beliebig viele Subscriber sie über einen Standard-Broker empfangen. In diesem Beitrag erfahren Sie, wie Part 14 der OPC-UA-Spezifikation PubSub strukturiert, worin sich JSON- und UADP-Kodierung unterscheiden, wie Topics und Metadaten organisiert sind und wie diese Nachrichten durch einen gewöhnlichen MQTT Broker fließen. Außerdem veröffentlichen und dekodieren Sie selbst eine JSON-NetworkMessage und sehen, wie PubSub-Daten in einen Unified Namespace (UNS) nach ISA-95 einfließen.

Warum OPC UA PubSub eingeführt hat

Klassisches OPC UA arbeitet nach dem Client/Server-Prinzip: Ein Client öffnet eine Session, legt Subscriptions an und empfängt Benachrichtigungen über diese Session. Das ist leistungsfähig, skaliert aber schlecht, wenn viele Konsumenten dieselben Daten wollen, und erfordert, dass Clients die Server direkt erreichen, was über Firewalls hinweg und in die Cloud umständlich ist. Unser Vergleich OPC UA vs. MQTT behandelt dieses Spannungsfeld ausführlich.

Part 14 der Spezifikation, eingeführt mit OPC UA 1.04 und in 1.05 erweitert, ergänzt ein Publish/Subscribe-Modell. Publisher senden Nachrichten, ohne zu wissen, wer sie empfängt; Subscriber filtern, was sie brauchen. Part 14 definiert zwei Familien von Transporten:

• Ohne Broker, etwa UDP-Multicast oder Ethernet (einschließlich TSN), für deterministische Kommunikation mit niedriger Latenz zwischen Steuerungen im selben Netzwerk.
• Broker-basiert, mit MQTT oder AMQP, um Daten an viele Konsumenten zu verteilen, über Netzwerkgrenzen hinweg und in die Cloud.

Dieser Artikel behandelt die zweite Familie mit MQTT als Transport.

Die PubSub-Bausteine

Part 14 führt ein Vokabular ein, das man kennen sollte, bevor man sich Nachrichten ansieht.

Der wichtige Gedanke: Die Payload ist kein beliebiges JSON, das sich jeder Hersteller selbst ausdenkt. Sie ist eine definierte Nachrichtenstruktur mit Sequenznummern, Metadatenversionen und einer Möglichkeit, Felder zu beschreiben, sodass ein generischer Subscriber Daten jedes konformen Publishers interpretieren kann.

JSON- vs. UADP-Kodierung

Part 14 definiert zwei Nachrichtenkodierungen, und beide können über MQTT transportiert werden.

Für die Cloud-Integration ist JSON die übliche Wahl, weil jede Programmiersprache es parsen kann und Sie Nachrichten direkt in einem Topic-Explorer prüfen können. UADP ist sinnvoll, wenn Bandbreite zählt oder wenn beide Enden OPC-UA-Stacks sind.

Aufbau einer JSON-NetworkMessage

Eine JSON-kodierte Datennachricht sieht so aus (Feldumfang zur besseren Lesbarkeit vereinfacht):

{
  "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
      }
    }
  ]
}

Die wichtigsten Felder:

• MessageType auf oberster Ebene unterscheidet Daten (ua-data) von Metadaten (ua-metadata) und, in neueren Versionen, Status- und andere Discovery-Nachrichten.
• PublisherId und DataSetWriterId identifizieren zusammen genau, um welchen Datensatz es sich handelt.
• SequenceNumber ermöglicht es Subscribern, Lücken und Duplikate zu erkennen.
• MetaDataVersion teilt dem Subscriber mit, welche Version der Feldbeschreibung gilt. Ändert sie sich, sollte der Subscriber neue Metadaten abrufen, bevor er Feldpositionen oder -typen vertraut.
• Key Frames vs. Delta Frames: Ein Key Frame enthält jedes Feld; ein Delta Frame nur die geänderten Felder, was den Datenverkehr bei sich langsam ändernden Daten reduziert.

Payload-Werte können in kompakterer oder ausführlicherer Form kodiert werden (die Spezifikation unterscheidet reversible und nicht reversible Kodierungen, und 1.05 verfeinert das). Die kompakte Form liefert reine Werte wie oben; die ausführliche Form umhüllt jeden Wert mit Typ, Status und Zeitstempeln, sodass er exakt in eine OPC-UA-Variant zurückdekodiert werden kann.

Topics und Metadaten auf dem Broker

In OPC UA 1.04 wird das MQTT-Topic (dort Queue Name genannt) einfach an jeder WriterGroup oder jedem DataSetWriter konfiguriert. OPC UA 1.05 ergänzt einen empfohlenen Standard-Topic-Baum, damit Subscriber Publisher auf vorhersehbare Weise finden können. Er hat diese Form:

<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

Das Präfix lautet standardmäßig opcua, ist aber konfigurierbar, was wichtig ist, wenn Sie einen Broker mit anderen Systemen teilen. Drei Konventionen sollten Sie befolgen:

1. Metadatennachrichten werden retained veröffentlicht, sodass ein später verbundener Subscriber die Feldbeschreibung sofort erhält.
2. Statusnachrichten sind retained und mit einem Last Will kombiniert, damit Subscriber wissen, ob ein Publisher online ist.
3. Die QoS ist pro WriterGroup konfigurierbar und wird auf MQTT-QoS 0, 1 oder 2 abgebildet (Best Effort, mindestens einmal, genau einmal).

Nichts davon erfordert besondere Broker-Unterstützung. Retained Messages, Last Will, Wildcards und QoS sind Standardfunktionen von MQTT, und genau deshalb hat Part 14 MQTT als Transport gewählt. Zur Auffrischung der Zustellgarantien lesen Sie MQTT-QoS-Stufen erklärt.

Wie Nachrichten durch einen Standard-Broker fließen

Aus Sicht des Brokers ist ein OPC-UA-PubSub-Publisher einfach ein weiterer MQTT-Client. Er verbindet sich, authentifiziert sich optional, setzt einen Last Will und veröffentlicht Payloads auf Topics. Der Broker parst die OPC-UA-Struktur nicht; er routet Bytes anhand von Topics und Subscriptions. Damit kann jeder standardkonforme MQTT Broker OPC-UA-PubSub-Verkehr transportieren, auch ein verwalteter MQTT-Broker wie ConnectMQ.

Ein Subscriber geht typischerweise so vor:

1. Er abonniert das Metadaten-Topic, erhält die retained Metadaten und baut eine Feldzuordnung auf.
2. Er abonniert das Daten-Topic oder eine Wildcard wie opcua/json/data/#.
3. Er prüft MetaDataVersion gegen die zwischengespeicherten Metadaten und SequenceNumber auf Lücken.
4. Er wendet Delta Frames auf den letzten bekannten Key Frame an, um den aktuellen Zustand zu halten.

Praxis: Eine JSON-NetworkMessage veröffentlichen und dekodieren

Der öffentliche Broker begrenzt Topics auf acht Ebenen, daher ersetzt das Beispiel das Präfix opcua durch ein eigenes, eindeutiges Präfix. Installieren Sie zuerst paho-mqtt 2.x mit pip install "paho-mqtt>=2.0".

Der Publisher sendet eine retained Metadatennachricht und anschließend alle zwei Sekunden einen Key Frame:

# pubsub_publisher.py - sendet JSON-Nachrichten im Stil von OPC UA Part 14 über MQTT
import json, random, time, uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/c3d9/json"          # steht für das Standardpräfix "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)

Der Subscriber speichert Metadaten zwischen, prüft Versionen und erkennt Sequenzlücken:

# pubsub_subscriber.py - dekodiert OPC-UA-JSON-NetworkMessages aus 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()

Starten Sie den Publisher und dann jederzeit den Subscriber. Weil die Metadaten retained sind, gibt der Subscriber zuerst die Feldliste und dann die Daten aus. Ersetzen Sie c3d9 durch eine eigene Zufallszeichenfolge.

Praktische Überlegungen

• Kombination mit dem Unified Namespace. OPC-UA-PubSub-Topics sind nach Publisher und WriterGroup gegliedert, nicht nach Werkshierarchie. Der UNS folgt dagegen der ISA-95-Anlagenhierarchie (Enterprise → Site → Area → Line → Cell), etwa acme/werk-stuttgart/zerspanung/linie-2/zelle-07/spindel. Viele Teams belassen den PubSub-Baum unverändert und betreiben einen kleinen Dienst, der ausgewählte Felder in den UNS erneut veröffentlicht, sodass MES, Historian und Dashboards sie am erwarteten Ort finden; siehe unseren Leitfaden zu ISA-95-MQTT-Topics und MQTT, Sparkplug B und der Unified Namespace.
• Nachrichtengröße. Key Frames mit vielen Feldern können groß werden. Prüfen Sie das Nachrichtengrößenlimit des Brokers für Ihren Plan und bevorzugen Sie bei breiten Datensätzen Delta Frames oder mehrere WriterGroups.
• Sicherheit. Über MQTT sichern Sie den Transport mit TLS und Broker-Zugangsdaten. Part 14 definiert außerdem Signierung und Verschlüsselung auf Nachrichtenebene mit einem Key Service, die vor allem mit UADP genutzt werden.
• Werkzeuge. Open-Source-Stacks wie open62541 enthalten PubSub mit MQTT-Unterstützung, und mehrere kommerzielle SPSen und Gateways können JSON nach Part 14 direkt veröffentlichen.

Für den größeren Zusammenhang, wo das in einer Werksarchitektur hingehört, lesen Sie unseren Leitfaden zu Industrial IoT mit MQTT.

Starten Sie mit ConnectMQ

Richten Sie Ihren OPC-UA-PubSub-Publisher auf einen Broker aus, den Sie nicht selbst betreiben müssen. Erstellen Sie ihn kostenlos unter app.connectmq.com im Serverless-Plan ohne Kreditkarte, verwenden Sie den generierten Benutzernamen, das Passwort und die Client ID mit TLS auf Port 8883 und prüfen Sie Ihre NetworkMessages im Live-Topic-Explorer des Portals. ConnectMQ unterstützt MQTT 3.1.1 und 5.0, Retained Messages und Last Will, also alles, worauf Part 14 aufbaut.

Häufig gestellte Fragen

Muss der MQTT Broker OPC UA verstehen, damit PubSub funktioniert?

Nein. Der Broker routet Nachrichten anhand von Topics und behandelt Payloads als Bytes. Jeder standardkonforme MQTT Broker kann OPC-UA-PubSub-Verkehr transportieren.

Sollte ich für OPC UA PubSub über MQTT JSON- oder UADP-Kodierung verwenden?

Verwenden Sie JSON, wenn IT- oder Cloud-Systeme die Daten konsumieren, weil es lesbar und leicht zu parsen ist. Verwenden Sie UADP, wenn die Bandbreite knapp ist oder beide Enden OPC-UA-Stacks sind.

Was ist der Unterschied zwischen einem Key Frame und einem Delta Frame?

Ein Key Frame enthält jedes Feld des Datensatzes; ein Delta Frame nur die geänderten Felder. Subscriber wenden Delta Frames auf den letzten Key Frame an, um den vollständigen Zustand zu halten.

Warum sollten Metadatennachrichten retained sein?

Damit ein Subscriber, der sich nach dem Start des Publishers verbindet, die Feldbeschreibung trotzdem sofort erhält, ohne darauf zu warten, dass der Publisher sie erneut sendet.