MQTT'de tekrarlanan mesajlar broker'ınızdaki bir hata değildir; teslimi "en az bir kez" vaat eden QoS 1'in belgelenmiş bir sonucudur. Bir ağ aksaklığı bir onayı yuttuğunda gönderen mesajı yeniden iletir ve tüketiciniz aynı mesajı iki kez görür. Bu sorun giderme rehberinde kopyaların tam olarak ne zaman oluştuğunu, DUP bayrağının size neyi söyleyip neyi söylemediğini ve bunları doğru şekilde ele alan idempotent bir Python tüketicisini nasıl yazacağınızı öğreneceksiniz.
QoS 1 Neden Kopya Üretir?
QoS 1, her aktarım adımında iki aşamalı bir el sıkışma kullanır: gönderen PUBLISH iletir, alıcı PUBACK ile yanıt verir. PUBACK gelene kadar gönderen mesajı tutar ve yeniden gönderir. MQTT QoS seviyeleri makalesi üç seviyeyi de ele alır; burada 1. seviyede neyin ters gittiğine odaklanıyoruz.
Alıcı mesajı almış ama gönderen onayı alamamışsa her seferinde bir kopya oluşur:
- Cihaz #42 numaralı okumayı broker'a publish eder.
- Broker onu alır, subscriber'lara iletir ve PUBACK gönderir.
- PUBACK cihaza ulaşmadan bağlantı kopar.
- Cihaz oturumuyla yeniden bağlanır ve #42'yi DUP bayrağı ayarlı olarak yeniden gönderir.
- Broker bunu yeni bir mesaj olarak ele alır ve yeniden iletir.
Aynı durum broker'dan subscriber'a giden adımda da yaşanır. Subscriber bir mesajı işlemiş ama PUBACK'i kaybolmuşsa, broker subscriber'ın oturumu devam ettiğinde mesajı yeniden gönderir.
Bir kopyanın her iki adımda da birbirinden bağımsız olarak gelebileceğini unutmayın; bu yüzden publish edilen tek bir mesaj zaman zaman bir tüketiciye ikiden fazla kez ulaşabilir.
DUP Bayrağı Size Ne Söyler?
Her PUBLISH paketinde bir DUP biti vardır. Gönderen, bir QoS 1 veya QoS 2 paketini yeniden iletirken bunu ayarlar. DUP ayarlı her mesajı atmak cazip gelebilir, ancak bu birkaç nedenle yanlıştır:
| Varsayım | Gerçek |
|---|---|
| DUP, bu mesaja zaten sahip olduğum anlamına gelir | Mutlaka değil. İlk kopya kaybolmuş olabilir ve yeniden deneme elinize geçen tek kopya olabilir |
| DUP yoksa mesaj yenidir | Hayır. Broker, bir publisher'ın yeniden denemesini giden adımda DUP olmadan yeni bir teslimat olarak iletir |
| Paket tanımlayıcıları kopyaları tespit edebilir | Yalnızca tek bir oturum ve tek bir iletim penceresi içinde. ID'ler sürekli yeniden kullanılır |
| QoS 2 her şeyi çözer | QoS 2 her adımda kopyaları önler, ancak uygulama düzeyindeki yeniden denemeler yine de kopya üretebilir |
DUP bayrağı loglama ve teşhis için kullanışlıdır. Bir tekilleştirme anahtarı değildir. Gerçek tekilleştirme, mesaja sizin koyduğunuz bir tanımlayıcı gerektirir.
Kopyaları Yeniden Üretmek
Gerçek yeniden teslim, kaybolan paketlere bağlıdır ve istendiğinde tetiklemek zordur. Tüketicinizi test etmenin güvenilir yolu kopyaları kendiniz eklemektir. broker.connectmq.com adresindeki herkese açık MQTT test broker'ına aynı mesaj ID'sini iki kez publish edin:
for n in 1 2; do
mosquitto_pub -h broker.connectmq.com -p 1883 -i dup-pub-t9v4 -q 1 \
-t connectmq/demo/t9v4/meter-3/energy \
-m '{"msgId":"meter-3-000042","kwh":0.25}'
done
Tekilleştirme yapmadan kwh değerlerini toplayan bir tüketici artık 0.25 yerine 0.5 bildirir. Bu, QoS 1 yeniden teslimin üretimde yol açtığı hata sınıfının ta kendisidir; yalnızca daha seyrek ve fark edilmesi daha zordur.
Idempotent Bir Tüketici Oluşturmak
Idempotent bir tüketici, bir mesajı ister bir kez ister beş kez işlesin aynı sonucu üretir. İki yaygın teknik vardır:
- Doğal idempotency. Etkiyi, tekrar zararsız olacak şekilde tasarlayın: "0.5 artır" yerine "sıcaklığı 21.5 yap"; satır eklemek yerine cihaz ve zaman damgasıyla anahtarlanmış bir satırı upsert edin.
- Açık tekilleştirme. Her mesaja benzersiz bir ID verin ve yakın zamanda işlenen ID'leri hatırlayın.
İşte sınırlı ve süreli bir önbellekle açık tekilleştirme kullanan eksiksiz bir paho-mqtt 2.x tüketicisi:
import json
import time
import uuid
from collections import OrderedDict
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
TOPIC = "connectmq/demo/t9v4/+/energy"
DEDUP_TTL = 600 # bir ID'nin hatırlanacağı saniye
DEDUP_MAX = 50_000 # bellek kullanımını sınırla
seen = OrderedDict() # msgId -> ilk görülme zamanı
totals = {}
def already_seen(msg_id):
now = time.time()
while seen:
oldest_id, ts = next(iter(seen.items()))
if now - ts > DEDUP_TTL or len(seen) > DEDUP_MAX:
seen.popitem(last=False)
else:
break
if msg_id in seen:
return True
seen[msg_id] = now
return False
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
try:
data = json.loads(msg.payload)
msg_id = data["msgId"]
except (ValueError, KeyError):
print("rejected message without msgId")
return
if already_seen(msg_id):
print(f"duplicate {msg_id} (dup flag={msg.dup}) ignored")
return
meter = msg.topic.split("/")[-2]
totals[meter] = totals.get(meter, 0) + data["kwh"]
print(f"{meter} total={totals[meter]:.2f} kWh")
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"dedup-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()
Bunu çalıştırın, ardından önceki bölümdeki çift publish döngüsünü çalıştırın. İlk mesaj toplamı günceller; ikincisi kopya olarak bildirilir ve yok sayılır.
İyi Bir Mesaj ID'si Seçmek
ID, orijinal publisher tarafından bir kez oluşturulmalı ve her yeniden denemede tekrar kullanılmalıdır. Seçenekler:
- Cihaz ID'si artı sıra numarası, örneğin
meter-3-000042. Kompakttır ve MQTT mesaj sıralaması kontrolleri için de kullanışlıdır. Yeniden başlatmadan sonra tekrarlanmaması için sayacı kalıcı olarak saklayın veya bir boot ID ekleyin. - Mesaj başına üretilen UUID. Basit ve global olarak benzersizdir; bedeli 36 bayttır.
- Cihaz aynı zaman damgasıyla asla iki okuma üretmiyorsa cihaz ID'si artı ölçüm zaman damgası.
ID'yi tüketicide veya bir ara katmanda üretmeyin; o noktada bir kopya yeni bir mesaj gibi görünür.
Tekilleştirmeyi Ölçeklendirmek
Yukarıdaki bellek içi önbellek tek bir tüketici süreci için çalışır. Daha büyük sistemler için:
- Veritabanı kısıtlamaları.
msg_idüzerindeki benzersiz bir indeks, kopyaları zararsız çakışmalara dönüştürür. Bir upsert ile (PostgreSQL'deINSERT ... ON CONFLICT DO NOTHING) çoğu zaman ihtiyacınız olan tek tekilleştirme budur. - Paylaşılan önbellek. Bir shared subscription'daki birden fazla tüketici aynı mesajın yeniden denemelerini alabiliyorsa, görülenler kümesi paylaşılmalıdır; örneğin TTL'li Redis'te.
- TTL boyutlandırma. Pencerenin yalnızca gerçekçi yeniden teslim gecikmelerini kapsaması gerekir: session expiry süreniz artı bir pay. Oturumlar o kadar uzun sürmedikçe kopyalar nadiren saatler arayla gelir.
Doğrudan QoS 2 mi Kullanmalısınız?
QoS 2, her adımda tam olarak bir kez teslimi garanti eden dört aşamalı bir el sıkışma ekler. Daha fazla gidiş-dönüşe ve broker durumuna mal olur; ayrıca zaman aşımına uğrayıp aynı okumayı uygulama düzeyinde yeni bir mesaj olarak yeniden publish eden bir cihaza karşı sizi korumaz. Kısıtlı planlarda hiç kullanılamayabilir; örneğin ücretsiz ConnectMQ Serverless planı QoS 1'e kadar destek verirken ücretli Serverless planı QoS 2'yi ekler. Broker'ların QoS'u nasıl bildirdiğini ve düşürdüğünü görmek için MQTT maksimum QoS yazısına bakın. Çoğu sistemde QoS 1 artı idempotent tüketiciler daha sağlam ve daha ucuz bir tasarımdır.
Sorun Giderme Kontrol Listesi
- Kopyalar yeniden bağlantılarla ilişkili mi? Client loglarında bağlantı kopmalarını ve keepalive zaman aşımlarını kontrol edin.
- Publisher'ınız QoS 1'in üzerine uygulama düzeyinde de yeniden deneme yapıyor mu? Bu, yeniden deneme yollarını ikiye katlar.
- İki client aynı client ID'yi paylaşıp birbirini düşürerek sürekli oturum devamına neden oluyor mu? Bkz. MQTT client sürekli bağlantıyı kesiyor.
- Tüketici onay vermekte yavaş mı? Callback içindeki uzun işlemler PUBACK'i geciktirir ve yeniden teslim penceresini genişletir.
- Her mesaj kaynakta oluşturulmuş bir ID taşıyor mu?
Kopyalar ve Broker Seçiminiz
ConnectMQ standart QoS 1 ve QoS 2 semantiğini uygular; bu yüzden yukarıdaki öneriler doğrudan geçerlidir. Client ID'ler verilen kimlik bilgilerine bağlı olduğundan, sıklıkla toplu yeniden teslime yol açan yinelenen client ID yeniden bağlanma döngüsünden kaçınmak daha kolaydır. Yönetilen bir MQTT broker oturum depolamayı ve yeniden denemeleri sizin için yönetir; tüketicilerinizdeki idempotency ise hangi broker olursa olsun yine sizin sorumluluğunuzdadır. Genel tablo için MQTT eğitimi iyi bir sonraki adımdır.
ConnectMQ ile Başlayın
app.connectmq.com adresinde, kredi kartı gerekmeden Serverless planında ücretsiz bir broker oluşturun. Tekilleştirme yapan tüketiciyi ConnectMQ kullanıcı adınız, parolanız ve client ID'nizle ona yönlendirin; ağ doğru davranmasa bile toplamlarınız doğru kalır.
Sık Sorulan Sorular
Neden aynı MQTT mesajını iki kez alıyorum?
QoS 1'de kaybolan bir onay, gönderenin mesajı yeniden göndermesine neden olur. Alıcı mesajı tekrar alır; bu, en az bir kez garantisi kapsamında izin verilen bir durumdur.
Kopyaları atmak için DUP bayrağına güvenebilir miyim?
Hayır. DUP bir adımdaki yeniden iletimi işaretler, ancak ilk kopya hiç ulaşmamış olabilir ve broker bayrağı subscriber'lara taşımaz.
QoS 0 kopya üretir mi?
Hayır, QoS 0 protokol tarafından asla yeniden iletilmez. Bunun yerine mesaj kaybedebilir.
Mesaj ID'lerini ne kadar süre hatırlamalıyım?
Yeniden bağlanma ve oturum devamı gecikmelerini kapsayacak kadar, genellikle birkaç dakika. Bunu session expiry aralığınıza bir güvenlik payı ekleyerek boyutlandırın.
