MQTT Üzerinden OPC UA PubSub: Mesajlar, Kodlamalar ve Topic'ler

MQTT üzerinden OPC UA PubSub, OPC UA'nın bilgi modelini MQTT'nin broker tabanlı dağıtımıyla birleştirir; böylece bir OPC UA cihazı veriyi bir kez yayımlar ve istenen sayıda abone onu standart bir broker üzerinden alabilir. Bu yazıda OPC UA spesifikasyonunun Part 14 bölümünün PubSub'ı nasıl yapılandırdığını, JSON ve UADP kodlamaları arasındaki farkı, topic'lerin ve meta verinin nasıl düzenlendiğini ve bu mesajların sıradan bir MQTT broker üzerinden nasıl aktığını öğreneceksiniz. Ayrıca bir JSON NetworkMessage'ı kendiniz yayımlayıp çözeceksiniz.

OPC UA Neden PubSub Ekledi?

Klasik OPC UA client/server'dır: bir client oturum açar, subscription'lar oluşturur ve bildirimleri bu oturum üzerinden alır. Güçlüdür, ama birçok tüketici aynı veriyi istediğinde kötü ölçeklenir ve client'ların sunuculara doğrudan ulaşmasını gerektirir; bu da güvenlik duvarları arasında ve buluta doğru zahmetlidir. OPC UA ve MQTT karşılaştırmamız bu gerilimi ayrıntılı olarak ele alıyor.

OPC UA 1.04 ile tanıtılan ve 1.05 ile genişletilen spesifikasyonun Part 14 bölümü bir publish/subscribe modeli ekler. Yayıncılar mesajları kimin alacağını bilmeden gönderir; aboneler ihtiyaç duyduklarını filtreler. Part 14 iki taşıma ailesi tanımlar:

• Broker'sız: aynı ağdaki kontrolörler arasında düşük gecikmeli, deterministik iletişim için UDP multicast veya Ethernet (TSN dahil) gibi.
• Broker tabanlı: veriyi birçok tüketiciye, ağlar arasında ve buluta dağıtmak için MQTT veya AMQP kullanan.

Bu yazı, taşıma katmanı olarak MQTT kullanan ikinci aileyle ilgilidir.

PubSub Yapı Taşları

Part 14, mesajlara bakmadan önce öğrenmeye değer bir sözlük getirir.

Önemli fikir şudur: payload, her üreticinin uydurduğu rastgele bir JSON değildir. Sıra numaraları, meta veri sürümleri ve alanları tanımlamanın bir yolu olan tanımlı bir mesaj yapısıdır; böylece genel bir abone, uyumlu herhangi bir yayıncıdan gelen veriyi yorumlayabilir.

JSON ve UADP Kodlaması

Part 14 iki mesaj kodlaması tanımlar ve ikisi de MQTT üzerinden taşınabilir.

Bulut entegrasyonu için JSON yaygın tercihtir, çünkü her dil onu ayrıştırabilir ve mesajları doğrudan bir topic explorer'da inceleyebilirsiniz. UADP, bant genişliği önemli olduğunda veya iki uç da OPC UA yığını olduğunda mantıklıdır.

Bir JSON NetworkMessage'ın Anatomisi

JSON kodlu bir veri mesajı şöyle görünür (alan seti okunabilirlik için sadeleştirilmiştir):

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

Anlaşılması gereken temel alanlar:

• Üst seviyedeki MessageType, veriyi (ua-data) meta veriden (ua-metadata) ve daha yeni sürümlerde durum ile diğer keşif mesajlarından ayırır.
• PublisherId ve DataSetWriterId birlikte bunun tam olarak hangi veri kümesi olduğunu tanımlar.
• SequenceNumber, abonelerin boşlukları ve kopyaları tespit etmesini sağlar.
• MetaDataVersion, aboneye alan tanımının hangi sürümünün geçerli olduğunu söyler. Değişirse abone, alan konumlarına veya tiplerine güvenmeden önce yeni meta veriyi almalıdır.
• Key frame ve delta frame: key frame her alanı taşır; delta frame yalnızca değişen alanları taşır, bu da yavaş değişen veriler için trafiği azaltır.

Payload değerleri daha kompakt veya daha ayrıntılı biçimlerde kodlanabilir (spesifikasyon geri döndürülebilir ve geri döndürülemez kodlamaları ayırır, 1.05 bunu inceltir). Kompakt biçim yukarıdaki gibi yalın değerler verir; ayrıntılı biçim ise her değeri tipi, durumu ve zaman damgalarıyla sarar, böylece tam bir OPC UA Variant'a geri çözülebilir.

Broker'da Topic'ler ve Meta Veri

OPC UA 1.04'te MQTT topic'i (queue name olarak adlandırılır) her WriterGroup veya DataSetWriter üzerinde basitçe yapılandırılır. OPC UA 1.05, abonelerin yayıncıları öngörülebilir bir şekilde keşfedebilmesi için önerilen bir varsayılan topic ağacı ekler. Şu biçimi izler:

<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

Önek varsayılan olarak opcua'dır ama yapılandırılabilir; bu, broker'ı başka sistemlerle paylaştığınızda önem taşır. İzlenmeye değer üç kural vardır:

1. Meta veri mesajları retained olarak yayımlanır; böylece sonradan bağlanan bir abone alan tanımını hemen alır.
2. Durum mesajları retained'dır ve bir Last Will ile eşleştirilir; böylece aboneler bir yayıncının çevrim içi olup olmadığını bilir.
3. QoS writer group başına yapılandırılabilir; MQTT QoS 0, 1 veya 2'ye (en iyi çaba, en az bir kez, tam olarak bir kez) karşılık gelir.

Bunların hiçbiri özel broker desteği gerektirmez. Retained message'lar, Last Will, wildcard'lar ve QoS standart MQTT özellikleridir ve Part 14'ün taşıma katmanı olarak MQTT'yi seçmesinin nedeni tam olarak budur. İletim garantilerini hatırlamak için MQTT QoS seviyeleri açıklandı yazımıza bakın.

Mesajlar Standart Bir Broker Üzerinden Nasıl Akar?

Broker'ın bakış açısından bir OPC UA PubSub yayıncısı sadece bir başka MQTT client'ıdır. Bağlanır, isteğe bağlı olarak kimlik doğrular, bir Last Will ayarlar ve topic'lere payload yayımlar. Broker OPC UA yapısını ayrıştırmaz; byte'ları topic'lere ve aboneliklere göre yönlendirir. Bu, standartlara uygun herhangi bir MQTT broker'ın, ConnectMQ gibi barındırılan bir MQTT broker dahil, OPC UA PubSub trafiğini taşıyabileceği anlamına gelir.

Bir abone tipik olarak:

1. Meta veri topic'ine abone olur, retained meta veriyi alır ve bir alan haritası oluşturur.
2. Veri topic'ine veya opcua/json/data/# gibi bir wildcard'a abone olur.
3. MetaDataVersion değerini önbellekteki meta veriyle, SequenceNumber değerini de boşluklar için kontrol eder.
4. Güncel durumu korumak için delta frame'leri bilinen son key frame'e uygular.

Uygulamalı: Bir JSON NetworkMessage Yayımlayın ve Çözün

Public broker topic'leri sekiz seviyeyle sınırlar; bu yüzden örnek, opcua önekini kendine ait benzersiz bir önekle değiştirir. Önce pip install "paho-mqtt>=2.0" ile paho-mqtt 2.x'i kurun.

Yayıncı retained bir meta veri mesajı gönderir, ardından her iki saniyede bir key frame gönderir:

# pubsub_publisher.py - MQTT üzerinden OPC UA Part 14 tarzı JSON mesajlar yayar
import json, random, time, uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/c3d9/json"          # varsayılan "opcua/json" yerine geçer
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)

Abone meta veriyi önbelleğe alır, sürümleri kontrol eder ve sıra boşluklarını tespit eder:

# pubsub_subscriber.py - MQTT'den gelen OPC UA JSON NetworkMessage'larını çözer
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()

Yayıncıyı çalıştırın, ardından aboneyi istediğiniz zaman başlatın. Meta veri retained olduğundan abone önce alan listesini, sonra veriyi yazdırır. c3d9 değerini kendi rastgele dizenizle değiştirin.

Pratik Hususlar

• Unified Namespace ile birlikte kullanım. OPC UA PubSub topic'leri tesis hiyerarşisine göre değil, yayıncıya ve writer group'a göre düzenlenir. Birçok ekip PubSub ağacını olduğu gibi tutar ve seçilen alanları ISA-95 tarzı bir namespace'e yeniden yayımlayan küçük bir servis çalıştırır; ISA-95 MQTT topic'leri rehberimize bakın.
• Mesaj boyutu. Çok alanlı key frame'ler büyüyebilir. Planınız için broker'ın mesaj boyutu sınırını kontrol edin ve geniş veri kümeleri için delta frame'leri veya birkaç writer group'u tercih edin.
• Güvenlik. MQTT üzerinden taşıma katmanını TLS ve broker kimlik bilgileriyle güvenceye alırsınız. Part 14 ayrıca, çoğunlukla UADP ile kullanılan, bir anahtar servisiyle mesaj seviyesinde imzalama ve şifreleme tanımlar.
• Araçlar. open62541 gibi açık kaynak yığınlar MQTT destekli PubSub içerir; birçok ticari PLC ve gateway de Part 14 JSON'u doğrudan yayımlayabilir.

Bunun bir tesis mimarisinde nereye oturduğuna dair daha geniş bağlam için endüstriyel IoT MQTT rehberimizi okuyun.

ConnectMQ ile Başlayın

OPC UA PubSub yayıncınızı işletmek zorunda olmadığınız bir broker'a yönlendirin. app.connectmq.com adresinde kredi kartı gerektirmeden Serverless planda ücretsiz bir broker oluşturun, 8883 portunda TLS ile oluşturulan kullanıcı adını, şifreyi ve client ID'yi kullanın ve NetworkMessage'larınızı portalın canlı topic explorer'ında inceleyin. ConnectMQ, Part 14'ün dayandığı her şeyi, yani MQTT 3.1.1 ve 5.0'ı, retained message'ları ve Last Will'i destekler.

Sık Sorulan Sorular

PubSub'ın çalışması için MQTT broker'ın OPC UA'yı anlaması gerekir mi?

Hayır. Broker mesajları topic'e göre yönlendirir ve payload'ları byte olarak ele alır. Standartlara uygun her MQTT broker OPC UA PubSub trafiğini taşıyabilir.

MQTT üzerinden OPC UA PubSub için JSON mı yoksa UADP kodlaması mı kullanmalıyım?

Veriyi IT veya bulut sistemleri tüketiyorsa JSON kullanın; okunabilir ve ayrıştırması kolaydır. Bant genişliği kısıtlıysa veya iki uç da OPC UA yığınıysa UADP kullanın.

Key frame ile delta frame arasındaki fark nedir?

Key frame veri kümesinin her alanını içerir; delta frame yalnızca değişen alanları içerir. Aboneler tam durumu korumak için delta frame'leri son key frame'e uygular.

Meta veri mesajları neden retained olmalı?

Yayıncı başladıktan sonra bağlanan bir abonenin, yayıncının onu yeniden göndermesini beklemeden alan tanımını hemen alabilmesi için.