MQTT Mesaj Sırasını Garanti Eder mi? Kendiniz Test Edin

MQTT mesaj sırasını yalnızca dar bir kapsamda garanti eder: tek bir publisher'dan, tek bir topic'e, tek bir QoS seviyesiyle gelen mesajlar her subscriber'a broker'ın onları aldığı sırayla teslim edilir. Topic'ler arasında, publisher'lar arasında ve paylaşılan aboneliklerde sıra garantisi yoktur — garanti edilen kapsamda bile QoS 1 yeniden gönderimleri geç gelen kopyalar üretebilir.

Bu makale spesifikasyonun tam olarak neyi vaat ettiğini, pratikte sıranın nerede bozulduğunu açıklıyor ve kendi broker'ınızda sırasız ve yinelenen mesajları ölçen iki kısa Python betiği sunuyor.

Spesifikasyon ne diyor

Hem MQTT 3.1.1 hem MQTT 5 (bölüm 4.6, "Message ordering") varsayılan olarak her topic'i bir Ordered Topic olarak tanımlar. Bir broker sıralı bir topic'teki mesajları iletirken, PUBLISH paketlerini subscriber'lara "aynı topic ve QoS için" herhangi bir istemciden aldığı sırayla göndermek zorundadır. İstemciler de onaylanmamış mesajları yeniden göndermek ve onayları sırayla göndermek zorundadır.

Spesifikasyon işin püf noktasını da belirtir. QoS 1 ile 1, 2, 3, 4 gönderen bir publisher, bir subscriber'ın 1, 2, 3, 2, 3, 4 görmesine yol açabilir: yeniden bağlanmanın ardından 2 ve 3'ün yeniden gönderilen kopyaları orijinallerinden sonra gelir. Her mesajın son kopyası sıralıdır, ancak bir kopya kendisinden sonraki mesajdan sonra görünebilir. Aynı anda yolda birden fazla mesaj yoksa bu olamaz.

MQTT QoS seviyeleri bu kuralların arkasındaki onay akışlarını açıklar.

Pratikte sıra nerede bozulur

• Yoldaki birden çok mesaj ve yeniden bağlanmalar. İstemciler ve broker'lar genellikle aynı anda çok sayıda onaylanmamış QoS 1/2 mesajına izin verir (MQTT 5'te limit Receive Maximum ile müzakere edilir). Bağlantı, yolda birkaç mesaj varken koparsa yeniden gönderim daha yeni mesajlardan sonra gelebilir.
• Paylaşılan abonelikler. $share/group/topic mesajları tüketiciler arasında dağıtır. Her tüketici kendi payını sıralı görebilir, ancak işleme paralel yürüdüğünden genel sonuç sırasızdır. Bkz. paylaşılan abonelikler.
• Birden çok publisher. Aynı cihazı raporlayan iki gateway, birbirinden bağımsız sıralanmış iki akış üretir. Broker bunları istediği şekilde iç içe geçirebilir.
• Bridge'ler ve cluster'lar. Bir bridge sadece başka bir istemcidir; tek bir bridge bağlantısı sırayı korur, broker'lar arasındaki birden çok yol korumaz. Cluster'lı broker'lar genellikle bir topic'te publisher başına sırayı korur, ancak broker'ınızın dokümantasyonunu kontrol edin.
• Kendi tüketiciniz. En yaygın neden — mesajlar sırayla gelir, ardından eşzamanlı işlenir (thread havuzları, async handler'lar, worker kuyrukları).

Test 1: broker'ınızda sırayı ölçün

Publisher mesajları topic ve publisher ID başına numaralandırır. Subscriber her (topic, publisher) için gördüğü en yüksek sıra numarasını izler ve geriye giden her şeyi, her kopyayı ve eksik olan her şeyi raporlar.

import os, sys, time
from collections import Counter, defaultdict
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
RUN = sys.argv[1]                      # publisher ile aynı çalıştırma kimliği
QOS = int(os.getenv("MQTT_QOS", "1"))
last = defaultdict(lambda: -1)         # (topic, publisher) -> en yüksek sıra numarası
seen = defaultdict(set)
stats = Counter()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"testmqtt/order/{RUN}/#", qos=QOS)

def on_message(client, userdata, msg):
    pub_id, seq = msg.payload.decode().split(":")
    key, seq = (msg.topic, pub_id), int(seq)
    stats["received"] += 1
    stats["dup_flag"] += msg.dup
    if seq in seen[key]:
        stats["duplicates"] += 1
        return
    seen[key].add(seq)
    if seq < last[key]:
        stats["out_of_order"] += 1
        print(f"OUT OF ORDER {key}: {seq} after {last[key]}")
    last[key] = max(last[key], seq)

def missing():
    return sum(last[k] + 1 - len(seen[k]) for k in seen)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
client.on_connect, client.on_message = on_connect, on_message
client.connect(HOST, 1883)
client.loop_start()
try:
    while True:
        time.sleep(5)
        print(dict(stats), "missing:", missing())
except KeyboardInterrupt:
    print("FINAL", dict(stats), "missing:", missing())

import os, sys
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
RUN, N = sys.argv[1], int(sys.argv[2]) if len(sys.argv) > 2 else 10000
PUB_ID = os.getenv("PUB_ID", "p1")
TOPICS = int(os.getenv("TOPICS", "4"))
QOS = int(os.getenv("MQTT_QOS", "1"))

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
if os.getenv("MQTT_USER"):
    pub.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
pub.max_inflight_messages_set(int(os.getenv("INFLIGHT", "20")))
pub.connect(HOST, 1883)
pub.loop_start()

seq, infos = [0] * TOPICS, []
for i in range(N):
    t = i % TOPICS
    infos.append(pub.publish(f"testmqtt/order/{RUN}/t{t}", f"{PUB_ID}:{seq[t]}", qos=QOS))
    seq[t] += 1
for info in infos:
    info.wait_for_publish(timeout=60)
pub.disconnect()
pub.loop_stop()
print(f"published {N} messages on {TOPICS} topics as {PUB_ID}")

pip install "paho-mqtt>=2"
python order_sub.py run42            # terminal 1, önce bunu başlatın
python order_pub.py run42 10000      # terminal 2

Sağlıklı bir broker ve kararlı bir ağda referans test out_of_order: 0, duplicates: 0 ve missing: 0 ile bitmelidir. Betiklere geçmeden önce, publisher çalışırken online MQTT istemcisinde testmqtt/order/# topic'ine abone olarak akışı canlı izleyebilirsiniz.

Test 2: uç durumları tetikleyin

Her seferinde tek bir değişkeni değiştirin ve referansla karşılaştırın:

Paylaşılan abonelik denemesi için subscribe çağrısını $share/... filtresiyle değiştirin; bir filtrenin topic'lerinizle eşleşip eşleşmediğini topic matcher ile kontrol edebilirsiniz. Kopyalar bir broker hatası değildir — bu, QoS 1'in işini yapmasıdır; yinelenen mesajlar tüketicileri nasıl idempotent yapacağınızı anlatır.

Sırayı kendi kodunuzda kaybetmeyin

"MQTT mesajlarımın sırasını bozdu" şikâyetlerinin çoğu tüketiciden kaynaklanır. Node.js'te async bir mesaj handler'ı bir önceki mesajı beklemez:

client.on("message", async (topic, payload) => {
  await saveToDb(topic, payload); // 2. mesaj 1. mesajdan önce bitebilir
});

Farklı topic'leri yine paralel işlerken topic başına sırayı korumak için işi topic başına zincirleyin:

const chains = new Map();
client.on("message", (topic, payload) => {
  const prev = chains.get(topic) ?? Promise.resolve();
  const next = prev.then(() => saveToDb(topic, payload)).catch(console.error);
  chains.set(topic, next);
});

Aynısı Python veya Java'daki thread havuzları ve broker'ın arkasındaki worker kuyrukları için de geçerlidir: işi topic'e veya cihaz ID'sine göre bölün, asla round-robin dağıtmayın.

Sırayı gözeten tasarım

• Sıralı kalması gereken olayları tek bir publisher'dan tek bir topic'e koyun — örneğin olay türü başına bir topic değil, site/device42/events.
• Payload'a bir sıra numarası veya kaynak zaman damgası ekleyin; böylece tüketiciler eski güncellemeleri atabilir ("son yazan kazanır") veya küçük pencereleri yeniden sıralayabilir.
• Handler'ları idempotent yapın, çünkü QoS 1 kopyalar teslim edecektir.
• Paylaşılan abonelikleri tasarım gereği sırasız kabul edin; sıra önemliyse cihaza göre bölün.
• Yalnızca son değerin önemli olduğu ve sıranın daha az önem taşıdığı "mevcut durum" için retained mesajları kullanın.

MQTT'nin sıra garantisi gerçektir ama dardır: tek publisher, tek topic, tek QoS. Sıra numaralı bir testle geçirilen on dakika, broker'ınızın ve ağınızın nasıl davrandığını gösterir; varyantlar ise tasarımınızın tam olarak nerede devreye girmesi gerektiğini ortaya koyar — genellikle payload'da bir sıra numarası ve her topic'i sırayla işleyen bir tüketiciyle.

Sıkça sorulan sorular