MQTT İstemci Performans Ayarı: Inflight, Batching ve Buffer'lar

"MQTT yavaş" sorunlarının çoğu broker'dan kaynaklanmaz. Her onayı bekleyen bir publisher, 100 ms'lik bir bağlantı üzerinde 20'lik bir inflight penceresi ya da dönmeden önce veritabanına yazan bir subscriber callback'i sorunun asıl kaynağıdır. İstemci tarafında MQTT performans ayarı birkaç ayara indirgenir: max inflight, nasıl publish ettiğiniz, gruplayıp gruplamadığınız, kuyruk boyutları, keep-alive ve TCP soketinin kendisi.

Bu rehber her ayarı açıklıyor, yaygın kütüphanelerdeki karşılığını gösteriyor ve etkisini öncesi ve sonrası ölçmeniz için bir betik veriyor.

Neden tavanı round-trip time belirler

QoS 1'de her mesaj bir PUBACK gerektirir. İstemci yalnızca N onaylanmamış mesaja izin veriyorsa, bağlantı başına throughput kabaca şununla sınırlıdır:

maks. mesaj/sn ≈ inflight penceresi ÷ round-trip time

20 inflight mesaj ve 50 ms round trip ile bu, broker ne kadar hızlı olursa olsun saniyede yaklaşık 400 mesajdır. Pencereyi 200'e çıkarın, tavan yaklaşık 4.000'e yükselir. 1 ms round trip'li bir LAN'da varsayılan pencere nadiren sınırdır; bölgeler arasında veya hücresel ağda neredeyse her zaman sınırdır.

QoS 0'da onay yoktur, dolayısıyla inflight penceresi geçerli değildir — orada sınırlar soket buffer'ları, CPU ve broker'ın istemci başına limitleridir. QoS 2 mesaj başına iki round trip gerektirir ve tavana daha erken ulaşır. Dengeyi görmek için QoS 0, 1 ve 2 benchmark'ı yazısına bakın.

Ayarlara genel bakış

Son sözü broker söyler. MQTT 5'te broker'ın CONNACK içindeki Receive Maximum değeri, istemcinin aynı anda kaç QoS 1/2 mesajı yolda tutabileceğini bildirir; uyumlu bir istemci yerelde ne ayarlarsanız ayarlayın bu değeri aşamaz. İstemci ayarını artırmak hiçbir şey değiştirmiyorsa bu değere bakın — akış kontrolü ve Receive Maximum değerin nasıl okunacağını anlatıyor.

1. Max inflight'ı artırın — dikkatle

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-tuned-pub")
client.max_inflight_messages_set(100)   # varsayılan 20
client.max_queued_messages_set(10000)   # sınırsız yerine belleği sınırla

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMaxInflight(100);               // varsayılan 10
opts.setKeepAliveInterval(30);

Daha büyük bir pencere bedava değildir. Her inflight mesaj onaylanana kadar istemci belleğinde kalır ve yeniden bağlanmadan sonra hepsi tekrar gönderilir — büyük bir pencere, kısa bir ağ kesintisini bir yinelenen mesaj dalgasına çevirir. Pencereyi adım adım büyütün ve throughput artmayı bıraktığında durun.

2. Asenkron publish edin

En büyük hata, bir sonrakini göndermeden önce her publish'in tamamlanmasını beklemektir:

for reading in readings:
    info = client.publish("testmqtt/tuning/data", reading, qos=1)
    info.wait_for_publish()             # her şeyi RTT'ye bağlar

paho-mqtt'de publish() mesajı zaten kuyruğa alır ve hemen döner. Öyle kalsın; tamamlanmayı on_publish içinde ya da MQTTMessageInfo nesnelerini saklayıp sonunda kontrol ederek takip edin.

MQTT.js 5'te publishAsync bir promise döner. Bir döngüde her birini await etmek aynı soruna yol açar; hepsini birden ateşlemek ise hiç sınır koymaz. Küçük bir pencere hem hız hem back-pressure sağlar:

import mqtt from "mqtt";

const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  clientId: `testmqtt-tune-${process.pid}`,
  keepalive: 30,
});

const WINDOW = 100;                      // kendi inflight sınırımız
const N = 5000;
const payload = Buffer.alloc(256, "x");
const pending = new Set();
const started = performance.now();

for (let i = 0; i < N; i++) {
  const p = client.publishAsync("testmqtt/tuning/js", payload, { qos: 1 })
    .finally(() => pending.delete(p));
  pending.add(p);
  if (pending.size >= WINDOW) await Promise.race(pending);
}
await Promise.all(pending);

const secs = (performance.now() - started) / 1000;
console.log(`${N} mesaj ${secs.toFixed(2)} sn'de = ${(N / secs).toFixed(0)} msg/s`);
await client.endAsync();

3. Küçük mesajları gruplayın

Her MQTT mesajı sabit bir header, topic adı ve — QoS 1'de — geri dönen 4 baytlık bir PUBACK taşır. 20 baytlık bir ölçüm için topic, verinin kendisinden büyük olabilir. Batching birçok ölçümü tek mesajda gönderir:

import json, time

buffer, last_flush = [], time.monotonic()

def add_reading(client, value):
    global last_flush
    buffer.append({"ts": time.time(), "v": value})
    if len(buffer) >= 50 or time.monotonic() - last_flush >= 1.0:
        client.publish("testmqtt/tuning/batch", json.dumps(buffer), qos=1)
        buffer.clear()
        last_flush = time.monotonic()

Grubu boyuta veya süreye göre boşaltın ki az veri üreten bir sensör de bir saniye içinde teslim etsin. Bedeller gerçektir: gecikme boşaltma aralığına kadar artar, kaybolan bir grup çok sayıda ölçümü götürür ve tüketicilerin dizileri anlaması gerekir. Gerçekte ne kadar overhead kazandığınızı görmek için paket boyutu hesaplayıcısını kullanın ve grupları broker'ın maksimum paket boyutunun epey altında tutun. MQTT 5'te topic alias'ları, payload formatını değiştirmeden uzun topic adlarını kısaltan daha hafif bir alternatiftir.

4. Buffer ve kuyrukları bilinçli boyutlandırın

Sınırsız bir giden kuyruk sorunları gizler: ağ üreticinizden yavaşsa bellek, süreç sonlandırılana kadar büyür. Sınırlı bir kuyruk bunu ele alabileceğiniz açık bir hataya dönüştürür. paho-mqtt'de max_queued_messages_set(n) sınırına ulaşıldığında publish() çağrısının döndürdüğü MQTTMessageInfo nesnesinde rc == MQTT_ERR_QUEUE_SIZE olur ve mesaj kuyruğa alınmaz — bunu loglayın ve mesajı atmaya, örneklemeye ya da yavaşlamaya karar verin.

Alıcı tarafta bunun karşılığı subscriber'ın işleme hızıdır. Callback'ler istemcinin ağ thread'inde veya event loop'unda çalışır; on_message meşgulken istemci soketi okumaz, PUBACK göndermez ve keep-alive'a yanıt vermez. İşi bir kuyruğa devredip hemen dönün:

import queue, threading

work = queue.Queue(maxsize=10000)

def on_message(client, userdata, msg):
    work.put_nowait((msg.topic, msg.payload))   # burada asla bloklama

def worker():
    while True:
        topic, payload = work.get()
        save_to_database(topic, payload)        # yavaş iş ağ thread'i dışında

threading.Thread(target=worker, daemon=True).start()

Kuyruk yine de doluyorsa tüketici gerçekten yavaştır — shared subscription'larla ölçekleyin veya yavaş tüketiciler yazısına bakın. Paho'nun Go istemcisinde SetOrderMatters(false), mesaj handler'larının sırayla değil paralel çalışmasını sağlar; bedeli handler'lar arasındaki sıralama garantisinin kaybıdır.

5. Keep-alive ve TCP_NODELAY

Keep-alive hiçbir şeyi hızlandırmaz. Kopmuş bir bağlantının ne kadar çabuk fark edileceğini belirler: broker, trafik olmadan yaklaşık 1,5 × keep-alive aralığı geçtikten sonra istemciyi düşürür. Bağlı cihazların çoğu için 30–60 saniye uygundur; pille çalışan bağlantılarda daha uzun kullanın. Kütüphanelerin çoğu başka paketler akarken PINGREQ göndermez. Ayrıntılar MQTT keep alive yazısında.

Nagle algoritması küçük mesajları geciktirebilir: işletim sistemi, önceki veriler onaylanmamışken küçük bir segmenti bekletir; karşı taraftaki gecikmeli ACK'lerle birleşince onlarca milisaniyelik gecikme sıçramaları oluşur. TCP_NODELAY ile kapatıldığında her paket hemen gönderilir. Kütüphanenizin bunu zaten yapıp yapmadığı kütüphaneye ve sürüme bağlıdır; kendiniz ayarlamanın zararı yoktur. paho-mqtt'de bunu on_socket_open callback'inde yapabilirsiniz:

import socket

def on_socket_open(client, userdata, sock):
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

client.on_socket_open = on_socket_open

Öncesini ve sonrasını ölçün

Ölçmeden ayar yapmak tahmin yürütmektir. Bu betik belirli bir inflight penceresiyle sabit sayıda QoS 1 mesajı publish eder ve onaylanma hızını raporlar:

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

N = 5000
INFLIGHT = int(sys.argv[1]) if len(sys.argv) > 1 else 20
acked, done = 0, threading.Event()

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1
    if acked == N:
        done.set()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-bench-{os.getpid()}")
client.max_inflight_messages_set(INFLIGHT)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
time.sleep(1)

start = time.perf_counter()
for _ in range(N):
    client.publish("testmqtt/tuning/bench", b"x" * 256, qos=1)
done.wait(timeout=120)
secs = time.perf_counter() - start
print(f"inflight={INFLIGHT}: {acked}/{N} onay {secs:.2f} sn'de = {acked / secs:.0f} msg/s")
client.loop_stop()
client.disconnect()

python inflight_bench.py 10 ile çalıştırın, ardından 20, 50, 100, 200 deneyin. Hız önce artmalı, sonra düzleşmelidir; eğrinin dirsek noktası o ağ yolu için doğru penceredir. Testi cihazlarınızın gerçekten bulunduğu yerden tekrarlayın — broker'ın yanında ölçülen bir sonuç başka bir kıtadaki cihaz hakkında pek bir şey söylemez. Betik çalışırken online istemciyi açın ve mesajların geldiğini doğrulamak için testmqtt/tuning/# topic'ine abone olun.

Çok sayıda istemcili daha büyük testler için özel araçlara geçin — throughput testi rehberi bu araçları ele alıyor.

Şu sırayla ayarlayın: her publish'i beklemeyi bırakın, eğri düzleşene kadar inflight penceresini büyütün, yalnızca mesaj başına overhead hâlâ baskınsa gruplayın, kuyruklarınızı sınırlayın ve callback'leri hızlı tutun. Her adım bir iki satır koddur ve benchmark betiği o adımın işe yarayıp yaramadığını size söyler.

Sıkça sorulan sorular