MQTT 5 ile MQTT İstek-Yanıt Deseni

MQTT istek-yanıt deseni, MQTT'nin kendisi bir publish/subscribe protokolü olsa da bir istemcinin soru sormasını ve belirli bir yanıt almasını sağlar. MQTT 5, bunu temiz ve birlikte çalışabilir hale getiren iki property ekledi: response topic ve correlation data. Bu rehberde Python ve Node.js ile çalışan RPC tarzı bir çağrı oluşturacak, onu herkese açık bir broker'da test edecek ve zaman aşımlarını ve hataları nasıl yöneteceğinizi öğreneceksiniz.

Neden MQTT Üzerinden İstek/Yanıt?

MQTT trafiğinin çoğu tek yönlüdür: sensörler publish eder, dashboard'lar subscribe olur. Ancak gerçek sistemlerin yanıtı olan sorulara da ihtiyacı vardır:

"Hangi firmware sürümünü çalıştırıyorsun?"
"40001 numaralı register'ı oku ve değerini söyle."
"Vana gerçekten hareket etti mi?"
"Bir öz test çalıştır ve sonucu bildir."

Her cihazda bir HTTP endpoint'i açabilirsiniz; ancak cihazlar çoğu zaman gelen bağlantıların imkânsız olduğu NAT veya hücresel ağların arkasında bulunur. Cihaz zaten giden bir MQTT bağlantısı tuttuğu için isteği ve yanıtı broker üzerinden yönlendirmek daha basittir ve tek bir güvenlik modeli korur. MQTT ve HTTP karşılaştırması, bu yalnızca giden bağlantıya dayalı modelin neden önemli olduğunu daha ayrıntılı ele alıyor.

MQTT 5'te MQTT İstek-Yanıt Deseni Nasıl Çalışır?

Desen, MQTT 5'te tanımlanan iki PUBLISH property'sini kullanır:

Property	Ayarlayan	Amaç
Response Topic	İstekte bulunan	Yanıt verenin cevabı publish etmesi gereken topic
Correlation Data	İstekte bulunan	Yanıt verenin değiştirmeden geri kopyaladığı opak baytlar

Akış şöyledir:

İstekte bulunan, kendi yanıt topic'ine subscribe olur; örneğin connectmq/demo/k3p8/rpc/replies/<requester-id>.
İsteği, Response Topic bu yanıt topic'ine ve Correlation Data benzersiz bir ID'ye ayarlanmış olarak yanıt verenin istek topic'ine publish eder.
Yanıt veren isteği alır, işi yapar ve Correlation Data'yı kopyalayarak sonucu Response Topic'e publish eder.
İstekte bulunan, gelen Correlation Data'yı bekleyen istekle eşleştirir ve tamamlar.

Broker istekleri hiç takip etmez. Yalnızca iki sıradan mesajı iletir. Onu "RPC" yapan her şey istemcilerde yaşar; desenin iyi ölçeklenmesinin ve herhangi bir MQTT 5 broker'ında çalışmasının nedeni budur.

İstek/Yanıt İçin Topic Tasarımı

İyi topic'ler deseni güvence altına almayı ve hata ayıklamayı kolaylaştırır:

İstekler: <prefix>/rpc/<device-id>/req; böylece her cihaz yalnızca kendi isteklerine subscribe olur.
Yanıtlar: <prefix>/rpc/replies/<requester-id>; böylece her çağıran yalnızca kendi yanıtlarını alır.
Birden fazla çağıran için asla paylaşılan bir yanıt topic'i kullanmayın. Her çağıran her yanıtı alır ve çoğunu atmak zorunda kalır.

İsimleri nasıl yapılandıracağınıza daha geniş bir açıdan bakmak isterseniz MQTT topic isimlendirme kuralları yazısına bakın.

Python ile Yanıt Veren

Bu yanıt veren, bir cihaz rolünü üstlenir. İstekleri dinler ve istekteki response topic ile correlation data'yı kullanarak bir sonuç döndürür. Herkese açık MQTT test broker'ında MQTT 5 ile paho-mqtt 2.x kullanır:

import json
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

BROKER = "broker.connectmq.com"
REQ_TOPIC = "connectmq/demo/k3p8/rpc/pump-07/req"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REQ_TOPIC, qos=1)
    print("responder ready")

def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    correlation = getattr(props, "CorrelationData", None)
    if not response_topic:
        print("request without response topic, ignoring")
        return

    request = json.loads(msg.payload)
    if request.get("method") == "getStatus":
        result = {"ok": True, "rpm": 1480, "running": True}
    else:
        result = {"ok": False, "error": "unknown method"}

    reply_props = Properties(PacketTypes.PUBLISH)
    if correlation is not None:
        reply_props.CorrelationData = correlation
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"responder-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Zaman Aşımlı Node.js İstemcisi

İstekte bulunan, bekleyen çağrıları correlation ID'ye göre anahtarlanmış bir map'te tutar ve zaman aşımı içinde yanıt almayan her çağrıyı reddeder. TLS üzerinden MQTT.js v5 kullanır:

const mqtt = require("mqtt");
const { randomUUID } = require("crypto");

const clientId = "requester-" + randomUUID().slice(0, 8);
const replyTopic = `connectmq/demo/k3p8/rpc/replies/${clientId}`;
const pending = new Map();

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
});

function call(deviceId, method, params = {}, timeoutMs = 5000) {
  const id = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(id);
      reject(new Error(`timeout calling ${method} on ${deviceId}`));
    }, timeoutMs);
    pending.set(id, { resolve, timer });
    client.publish(
      `connectmq/demo/k3p8/rpc/${deviceId}/req`,
      JSON.stringify({ method, params }),
      {
        qos: 1,
        properties: {
          responseTopic: replyTopic,
          correlationData: Buffer.from(id),
          messageExpiryInterval: Math.ceil(timeoutMs / 1000),
        },
      }
    );
  });
}

client.on("message", (topic, payload, packet) => {
  const corr = packet.properties && packet.properties.correlationData;
  if (!corr) return;
  const id = corr.toString();
  const entry = pending.get(id);
  if (!entry) return; // geç gelen veya bilinmeyen yanıt
  clearTimeout(entry.timer);
  pending.delete(id);
  entry.resolve(JSON.parse(payload.toString()));
});

client.on("connect", () => {
  client.subscribe(replyTopic, { qos: 1 }, async (err) => {
    if (err) return console.error(err);
    try {
      const status = await call("pump-07", "getStatus");
      console.log("reply:", status);
    } catch (e) {
      console.error(e.message);
    } finally {
      client.end();
    }
  });
});

Python yanıt vereni başlatın, ardından Node.js istemcisini çalıştırın. reply: { ok: true, rpm: 1480, running: true } çıktısını görmelisiniz. Zaman aşımı yolunu görmek için yanıt vereni durdurup istemciyi yeniden çalıştırın.

İsteğin zaman aşımına eşit bir mesaj sona erme süresi de taşıdığına dikkat edin. Cihaz kalıcı bir oturumla çevrimdışıysa, istek dakikalar sonra çalıştırılmak yerine çağıran vazgeçtiği anda atılır. Ayrıntılar MQTT mesaj sona erme süresi makalemizde yer alıyor.

İstek/Yanıt Tasarım Kontrol Listesi

İlk isteği göndermeden önce yanıt topic'ine subscribe olun; aksi halde hızlı bir yanıtı kaçırabilirsiniz.
Her istek için UUID gibi benzersiz bir correlation ID kullanın.
Her zaman bir zaman aşımı uygulayın ve bekleyen kayıtları temizleyin.
Eskimiş çağrıların asla çalıştırılmaması için isteklerde mesaj sona erme süresi ayarlayın.
Sessiz kalmak yerine yapılandırılmış hatalar (ok: false, hata kodu) döndürün.
QoS 1 bir isteği iki kez iletebileceği için eylemleri mümkün olduğunca idempotent yapın. Bkz. MQTT tekrarlanan mesajlar.
İstekleri veya yanıtları retain etmeyin.

Yanıt Payload'ını Tasarlamak

Taşıma katmanı sözleşmenin yalnızca yarısıdır; birden fazla ekip aynı cihazları çağırmaya başladığında yanıt formatı da en az o kadar önemlidir. Küçük ve tutarlı bir zarf, çağıranları basit tutar:

{"ok": false, "code": "BUSY", "message": "self-test already running", "retryAfter": 10}

Pratikte iyi işleyen bazı konvansiyonlar:

Hata durumunda bile her zaman yanıt verin. Sessiz bir yanıt veren, çevrimdışı olandan ayırt edilemez ve çağıran yalnızca zaman aşımına uğrayabilir.
Makine tarafından okunabilir kodlar kullanın; örneğin BUSY, INVALID_PARAMS veya UNKNOWN_METHOD ve loglar için insan tarafından okunabilir bir mesaj.
Method'larınızı sürümleyin. getStatus.v2 veya bir version alanı, eski çağıranları bozmadan yanıt yapısını değiştirmenize olanak tanır.
Kabulü tamamlanmadan ayırın. Bir iki saniyeden uzun süren her şey için hemen {"ok": true, "accepted": true, "jobId": "..."} ile yanıt verin ve ilerlemeyi ve nihai sonucu bir iş topic'ine publish edin. Böylece çağıranın zaman aşımı yalnızca kabulü kapsar.
Content type'ı belirtin. İsteklerde ve yanıtlarda MQTT 5 Content Type property'sini application/json olarak ayarlamak, genel araçların bunları doğru göstermesine yardımcı olur.

MQTT 3.1.1 İstemcileriyle Çalışmak

Bazı cihazlar yalnızca MQTT 3.1.1 konuşuyorsa property'ler kullanılamaz. Yaygın çözüm aynı bilgiyi payload'a koymaktır; örneğin {"replyTo": "...", "corrId": "...", "method": "getStatus"}. Bu çalışır, ancak her tarafın alan adları üzerinde anlaşması gerekir; MQTT 5 property'leri ise standarttır ve payload'ı temiz tutar. Karma filolar, eski firmware emekliye ayrılana kadar çoğu zaman ikisini birden çalıştırır.

Sorun Giderme

Hiç yanıt gelmiyor. Yanıt verenin Response Topic'i gerçekten gördüğünü kontrol edin. Taraflardan biri MQTT 3.1.1 ile bağlanırsa property'ler düşer. Ayrıca istekte bulunanın tam olarak doğru yanıt topic'ine subscribe olduğunu doğrulayın.

Yanıtlar yanlış çağırana gidiyor. Yanıt topic'iniz muhtemelen paylaşılıyordur. İstekte bulunanın client ID'sini topic'e ekleyin.

Correlation data eşleşmiyor. İkili veridir. Baytları karşılaştırın ya da örneklerde UTF-8 string'lerle yapıldığı gibi her iki tarafı tutarlı şekilde çözümleyin.

Çağrılar yük altında askıda kalıyor. Tek thread'li bir yanıt veren bir seferde bir isteği işler. Uzun işlemler "accepted" durumuyla yanıt vermeli ve nihai sonucu daha sonra publish etmelidir.

ConnectMQ'da RPC Çalıştırmak

ConnectMQ, MQTT 5.0 response topic, correlation data ve message expiry'yi destekler; bu yüzden yukarıdaki kod kendi broker'ınızda değişiklik yapmadan çalışır. Host'u kendi broker'ınızla değiştirin ve portaldan aldığınız username: "YOUR_USERNAME", password: "YOUR_PASSWORD" ve clientId: "YOUR_CLIENT_ID" değerlerini ekleyin. Barındırılan bir MQTT broker operasyonel tarafı ortadan kaldırır; böylece istek sözleşmesine odaklanabilirsiniz. Büyük resim için MQTT eğitimiyle başlayın.

ConnectMQ ile Başlayın

app.connectmq.com adresinde kredi kartı gerekmeden Serverless planda ücretsiz bir broker oluşturun. Yanıt vereni bir cihazda veya dizüstü bilgisayarda çalıştırın, bir betikten çağırın ve isteklerin ve yanıtların ConnectMQ portalının canlı topic gezgininde akışını izleyin.

Sık Sorulan Sorular

Broker istekleri yanıtlarla eşleştirir mi?

Hayır. Broker birbirinden bağımsız iki mesajı iletir. İstekte bulunan, kendi seçtiği correlation data'yı kullanarak bunları eşleştirir.

Correlation data zorunlu mu?

Kesin olarak değil; ancak onsuz, bekleyen birden fazla isteği olan bir istemci hangi yanıtın hangi çağrıya ait olduğunu ayırt edemez. Her zaman ayarlayın.

İstek/yanıt için QoS 0 kullanabilir miyim?

Kullanabilirsiniz, ancak bu durumda kaybolan istekler yalnızca zaman aşımı olarak ortaya çıkar. Idempotent handler'larla QoS 1 olağan tercihtir.

MQTT istek/yanıt, HTTP API'lerin yerini alır mı?

NAT arkasındaki cihaz komutları ve sorguları için çoğu zaman daha uygundur. Büyük dosya aktarımları veya herkese açık API'ler için HTTP genellikle daha basittir.

Response Topic ve Correlation Data ile MQTT İstek/Yanıt Deseni