MQTT mesaj sıralaması garanti edilir, ancak birçok geliştiricinin hiç okumadığı belirli sınırlar içinde. Tek bir publisher'dan, tek bir topic'te, tek bir QoS seviyesinde gönderilen mesajlar sırayla ulaşır; bunun ötesindeki neredeyse hiçbir şey vaat edilmez. Bu makale spesifikasyonun tam olarak neyi garanti ettiğini, mesajların sırasız göründüğü yaygın durumları ve tüketicilerinizin bunu tespit edip ele alabilmesi için sıra numaralarını nasıl ekleyeceğinizi açıklıyor.
MQTT Spesifikasyonu Neyi Garanti Eder?
Hem MQTT 3.1.1 hem de MQTT 5 bir "Ordered Topic" (sıralı topic) tanımlar. Varsayılan olarak broker her topic'i sıralı kabul etmek zorundadır ve bu da şu garantiye yol açar:
Bir client tarafından bir topic'e aynı QoS seviyesinde publish edilen mesajlar, her subscriber'a publish edildikleri sırayla teslim edilir.
Spesifikasyon ayrıca, bir client yeniden bağlandıktan sonra onaylanmamış QoS 1 ve QoS 2 mesajlarını yeniden gönderirken bunu orijinal sırayla yapmasını ve onayların paketlerin alındığı sırayla gönderilmesini şart koşar.
Bu garantiyi dikkatle okuduğunuzda boşluklar belirginleşir. Şunlar hakkında hiçbir şey söylemez:
- Aynı topic'teki farklı publisher'lardan gelen mesajlar.
- Bir publisher'dan farklı topic'lere giden mesajlar.
- Bir publisher'dan aynı topic'e ama farklı QoS seviyelerinde gelen mesajlar.
- Client kütüphanesi mesajı teslim ettikten sonra uygulamanızın ne yaptığı.
MQTT Mesaj Sıralaması Nerede Bozulur?
| Durum | Sıralama garanti mi? | Neden |
|---|---|---|
| Bir publisher, bir topic, aynı QoS | Evet | Ordered Topic kuralı |
| Bir publisher, iki topic | Hayır | Her topic bağımsız olarak sıralıdır |
| İki publisher, bir topic | Hayır | Global bir saat yoktur; broker'a varış sırası geçerlidir |
| Aynı topic, karışık QoS 0 ve QoS 1 | Hayır | Farklı teslim yolları ve yeniden denemeler |
| Kopan bağlantı üzerinden QoS 0 | Sıra korunur ama boşluklar olur | Kaybolan mesajlar yeniden gönderilmez |
| Yeniden teslimli QoS 1 | Kopyalar görülebilir | Yeniden gönderilen bir mesaj, zaten önünde olduğu yeni mesajların ardından gelebilir |
| Shared subscription | Global sıra yok | Mesajlar tüketicilere dağıtılır |
| Çok iş parçacıklı tüketici | Hayır | Worker havuzunuz işleme sırasını değiştirir |
Bunlardan ikisi daha yakından bakmayı hak ediyor.
Shared subscription'lar. $share/group/topic ile broker, yük dengeleme için mesajları birkaç tüketiciye dağıtır. Her tüketici bir alt kümeyi görür ve paralel işler; bu yüzden anlamlı bir global sıra yoktur. Cihaz başına sıralama önemliyse, her cihazın mesajları hep aynı tüketiciye ulaşacak şekilde tasarlayın ya da sıralamayı sonraki aşamada yeniden kurun. MQTT shared subscription'lar rehberimiz bu ödünleşimi ele alır.
Yeniden teslim ve kopyalar. QoS 1 en az bir kez (at-least-once) teslim demektir. Bir onay kaybolursa gönderen, DUP bayrağı ayarlı olarak yeniden iletir. Bu durumda bir subscriber, 6 ve 7'yi zaten işledikten sonra 5 numaralı mesajı yeniden görebilir. Bu gerçek bir sıra bozulması değil, bir kopyadır; ancak basit bir kod için zaman geriye gitmiş gibi görünür. Çözüm, MQTT'de tekrarlanan mesajları ele alma yazısında anlatılıyor.
Herkese Açık Bir Broker'da Sıralamayı Göstermek
En hızlı deney, broker.connectmq.com adresindeki herkese açık MQTT test broker'ını kullanır. Client başına saniyede 10 mesaj sınırına dikkat edin; aşağıdaki publisher bu yüzden kendini yavaşlatır.
Subscriber:
mosquitto_sub -h broker.connectmq.com -p 1883 -i ord-sub-h5d3 -q 1 \
-t 'connectmq/demo/h5d3/#' -v
Publisher; tek topic, QoS 1, numaralı mesajlar:
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 1883 -i ord-pub-h5d3 -q 1 \
-t connectmq/demo/h5d3/seq -m "$i"
sleep 0.2
done
1'den 20'ye kadar sırayla göreceksiniz. Her mosquitto_pub çağrısının ayrı bir bağlantı olduğunu unutmayın; burada sırayı koruyan şey ardışık döngüdür. Şimdi iki farklı topic'te, seq-a ve seq-b, iki döngüyü paralel çalıştırın. Her topic kendi içinde sıralı kalır, ancak iki topic'in satırları öngörülemez şekilde iç içe geçer. Bir sorun yoktur: topic'ler arası sıralama sözleşmenin parçası değildir.
Sıra Numaraları Eklemek
Sıralama uygulamanız için önemliyse onu payload'da taşıyın. Publisher başına monoton artan bir sıra numarası ve bir zaman damgası, tüketicilerin boşlukları, kopyaları ve eskimiş mesajları tespit etmesini sağlar. İşte eksiksiz bir paho-mqtt 2.x publisher ve tüketici çifti:
import json
import time
import uuid
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
TOPIC = "connectmq/demo/h5d3/pump-07/telemetry"
last_seq = {} # cihaz -> görülen en yüksek sıra numarası
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
data = json.loads(msg.payload)
dev, seq = data["device"], data["seq"]
prev = last_seq.get(dev)
if prev is not None and seq <= prev:
print(f"stale or duplicate seq={seq} (last={prev}), dropped")
return
if prev is not None and seq > prev + 1:
print(f"gap: missing {prev + 1}..{seq - 1}")
last_seq[dev] = seq
print(f"apply seq={seq} value={data['value']}")
sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"ord-c-{uuid.uuid4().hex[:6]}")
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(BROKER, 1883, keepalive=30)
sub.loop_start()
time.sleep(1)
pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"ord-p-{uuid.uuid4().hex[:6]}")
pub.connect(BROKER, 1883, keepalive=30)
pub.loop_start()
for seq in [1, 2, 3, 5, 4, 6]: # 4 bilerek geç gönderiliyor
body = {"device": "pump-07", "seq": seq, "ts": time.time(), "value": 20 + seq}
pub.publish(TOPIC, json.dumps(body), qos=1).wait_for_publish()
time.sleep(0.2)
time.sleep(2)
for c in (pub, sub):
c.loop_stop()
c.disconnect()
Tüketici 1, 2 ve 3'ü uygular, 5'ten önce bir boşluk bildirir, geç gelen 4'ü eskimiş olarak atar ve ardından 6'yı uygular. Geç gelen mesajların atılması, tamponlanması veya kabul edilmesi veriye bağlıdır; bu da bizi stratejiye getiriyor.
Bir Sıralama Stratejisi Seçmek
- Son değer kazanır (durum). Sıcaklık gibi okumalarda elinizdekinden eski olan her şeyi atın. Yukarıdaki kod bunu yapar.
- Yeniden sıralama tamponu (olay akışları). Mesajları kısa süre tutup sıra numarasına göre serbest bırakın ve bir zaman aşımından sonra boşlukları beklemeyi bırakın. Sayaçlar gibi her olayın önemli olduğu durumlarda kullanışlıdır.
- Idempotent upsert'ler (depolama). Her mesajı cihaz ve sıra numarası ya da zaman damgasıyla anahtarlayarak yazın. Varış sırası ne olursa olsun veritabanı doğru sonuçlanır.
- Sıralı her akış için bir topic. Komutların sırayla uygulanması gerekiyorsa onları tek bir client'tan, tek bir topic'te ve tek bir QoS ile publish edin.
- Bir akışta QoS karıştırmaktan kaçının. Topic başına tek bir QoS seçin.
Yeniden Bağlantılar ve Çevrimdışı Dönemlerde Sıralama
Sıralama varsayımlarının en çok sınandığı yer yeniden bağlantılardır. Kalıcı oturumu olan bir subscriber geri döndüğünde broker önce iletimde olan ve onaylanmamış mesajları yeniden gönderir, ardından client çevrimdışıyken kuyruğa alınan mesajları teslim eder, sonra canlı trafiğe devam eder. Bir publisher'dan gelen tek bir topic içinde sıra yine korunur, ancak tüketici aniden bir grup eski mesaj ve ardından güncel mesajlar alır.
Burada iki şey yardımcı olur. Birincisi, cihaz başına işlemeyi ardışık tutun; böylece bu ani yük paralel worker'lar tarafından yarıştırılmak yerine sırayla uygulanır. İkincisi, yetişme verisini nasıl sunacağınıza karar vermek için payload'daki zaman damgalarını kullanın: bir dashboard doğrudan en son değere atlamak isteyebilirken bir historian her okumayı orijinal zamanıyla saklamalıdır. Kuyruktaki mesajlar yalnızca sınırlı bir süre için faydalıysa, MQTT 5 message expiry çok eski mesajları bu ani yükün tamamen dışında tutar.
Publish tarafında, ağ kesintisi sırasında okumaları yerel olarak tamponlayan cihazlar, canlı publish'e dönmeden önce tamponu sıra numarasına göre ve normal trafikle aynı topic ve QoS üzerinden göndermelidir.
Sıralama Tuzakları ve Sorun Giderme
Yeniden başlatmadan sonra sıra numarası sıfırlanıyor. seq 1'den yeniden başlayan bir cihaz eskimiş veri gönderiyormuş gibi görünür. Sıra numarasıyla birlikte bir boot ID veya epoch sayacı ekleyin ya da sayacı flash bellekte saklayın.
Saate dayalı sıralama başarısız oluyor. Cihaz saatleri kayar ve bazen 1970'e sıfırlanır. Sıralama için sıra numaralarını, gösterim için zaman damgalarını tercih edin.
Sıra uygulamanızın içinde kayboluyor. Tüketicideki async handler'lar, thread havuzları ve kuyruklar sırayı serbestçe değiştirir. Her cihazın akışını tek bir worker'da işleyin veya cihaz ID'sine göre bölümleyin.
İletimdeki pencerenin etkileri. MQTT 5'in receive maximum ayarı, aynı anda birden fazla QoS 1 mesajının iletimde olmasına izin verir. Hat üzerindeki sıralama yine korunur, ancak paket kaybı altındaki yeniden denemeler kopyaları daha olası hale getirir. Ayar için MQTT receive maximum makalesine bakın.
Yönetilen Bir Broker'da Sıralama
ConnectMQ, MQTT 3.1.1 ve 5.0 client'ları için topic bazında MQTT sıralama kurallarına uyar. Ücretli Serverless planındaki shared subscription'lar, tüketiciler için yatay ölçeklenme sağlar; yukarıda anlatılan olağan sıralama ödünleşimi de buna dahildir. Barındırılan bir MQTT broker kullanmak protokol garantilerini değiştirmez, ancak operasyonel yükü ortadan kaldırarak tüketici tasarımına odaklanmanızı sağlar. Protokolün daha geniş bağlamı için MQTT eğitimine göz atın.
ConnectMQ ile Başlayın
app.connectmq.com adresinde, kredi kartı gerekmeden ücretsiz Serverless planında bir broker oluşturun. Sıra numarası örneğini kendi ConnectMQ kimlik bilgilerinizle çalıştırın, akışı portalın canlı topic gezgininde izleyin ve ardından filonuzla birlikte ölçeklenen bir bulut MQTT broker'ında üretime taşıyın.
Sık Sorulan Sorular
MQTT mesaj sırasını garanti eder mi?
Evet, bir client'tan bir topic'e aynı QoS ile gönderilen mesajlar için. Topic'ler, publisher'lar veya karışık QoS seviyeleri arasında bir garanti yoktur.
QoS 2 sıralamayı garanti eder mi?
QoS 2, her aktarım adımında tam olarak bir kez (exactly-once) teslimi garanti eder. Sıralama yine aynı topic ve publisher bazlı kurala uyar.
Neden yeni bir mesajdan sonra daha eski bir mesaj görüyorum?
Genellikle bir QoS 1 yeniden teslimi, ikinci bir publisher veya tüketicinizdeki paralel işleme nedeniyle. Sıra numaraları hangisi olduğunu anlamayı kolaylaştırır.
Shared subscription'lar sıralı mıdır?
Global olarak değil. Her mesaj grubun bir üyesine gider ve üyeler paralel olarak işler.
