Optimiser les performances d'un client MQTT : inflight, batching et buffers

La plupart des problèmes de « MQTT lent » ne viennent pas du broker. Il s'agit d'un publisher qui attend chaque acquittement, d'une fenêtre inflight de 20 sur un lien à 100 ms, ou d'un callback d'abonné qui écrit en base de données avant de rendre la main. Optimiser les performances d'un client MQTT revient à une poignée de réglages : max inflight, la manière de publier, le regroupement, la taille des files, le keep-alive et le socket TCP lui-même.

Ce guide explique chaque réglage, montre sa configuration dans les bibliothèques courantes et fournit un script pour mesurer l'effet avant et après.

Pourquoi le round-trip time fixe le plafond

En QoS 1, chaque message nécessite un PUBACK. Si le client n'autorise que N messages non acquittés, le débit par connexion est limité à environ :

max. messages/s ≈ fenêtre inflight ÷ round-trip time

Avec 20 messages inflight et un aller-retour de 50 ms, cela fait environ 400 messages par seconde, quelle que soit la rapidité du broker. Passez la fenêtre à 200 et le plafond monte à environ 4 000. Sur un LAN avec 1 ms d'aller-retour, la fenêtre par défaut est rarement la limite ; entre régions ou sur réseau cellulaire, elle l'est presque toujours.

QoS 0 n'a pas d'acquittements, la fenêtre inflight ne s'applique donc pas : les limites sont alors les buffers du socket, le CPU et les limites par client du broker. QoS 2 nécessite deux allers-retours par message et atteint donc le plafond plus tôt. Voir les benchmarks QoS 0 vs 1 vs 2 pour le compromis.

Les réglages en un coup d'œil

Le broker a le dernier mot. En MQTT 5, le Receive Maximum du broker dans le CONNACK indique au client combien de messages QoS 1/2 il peut avoir en vol ; un client conforme ne doit pas le dépasser, quel que soit votre réglage local. Si augmenter le réglage client ne change rien, vérifiez cette valeur — contrôle de flux et Receive Maximum explique comment la lire.

1. Augmenter le max inflight, prudemment

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-tuned-pub")
client.max_inflight_messages_set(100)   # défaut 20
client.max_queued_messages_set(10000)   # borne la mémoire au lieu d'illimitée

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMaxInflight(100);               // défaut 10
opts.setKeepAliveInterval(30);

Une fenêtre plus grande n'est pas gratuite. Chaque message inflight reste en mémoire côté client jusqu'à son acquittement, et après une reconnexion tous sont renvoyés : une grande fenêtre transforme une micro-coupure réseau en rafale de messages dupliqués. Augmentez-la par paliers et arrêtez-vous quand le débit ne progresse plus.

2. Publier en asynchrone

La plus grosse erreur consiste à attendre la fin de chaque publication avant d'envoyer la suivante :

for reading in readings:
    info = client.publish("testmqtt/tuning/data", reading, qos=1)
    info.wait_for_publish()             # sérialise tout sur le RTT

Dans paho-mqtt, publish() met déjà le message en file et rend la main immédiatement. Laissez-le faire et suivez la fin des envois dans on_publish, ou conservez les objets MQTTMessageInfo pour les vérifier à la fin.

Dans MQTT.js 5, publishAsync renvoie une promesse. Attendre chacune dans une boucle pose le même problème ; les lancer toutes d'un coup n'a aucune limite. Une petite fenêtre vous donne à la fois la vitesse et le back-pressure :

import mqtt from "mqtt";

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

const WINDOW = 100;                      // notre propre limite inflight
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} msgs en ${secs.toFixed(2)} s = ${(N / secs).toFixed(0)} msg/s`);
await client.endAsync();

3. Regrouper les petits messages

Chaque message MQTT transporte un en-tête fixe, le nom du topic et — en QoS 1 — un PUBACK de 4 octets au retour. Pour une mesure de 20 octets, le topic peut être plus gros que les données. Le batching envoie un seul message contenant de nombreuses mesures :

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()

Videz le lot sur taille ou sur durée, pour qu'un capteur peu actif livre quand même en moins d'une seconde. Les compromis sont réels : la latence augmente jusqu'à l'intervalle de vidage, un lot perdu emporte de nombreuses mesures et les consommateurs doivent comprendre les tableaux. Utilisez le calculateur de taille de paquet pour voir l'overhead réellement économisé, et gardez les lots bien en dessous de la taille maximale de paquet du broker. En MQTT 5, les topic aliases sont une alternative plus légère qui raccourcit les noms de topic longs sans changer le format du payload.

4. Dimensionner buffers et files volontairement

Une file sortante illimitée masque les problèmes : quand le réseau est plus lent que votre producteur, la mémoire grossit jusqu'à ce que le processus soit tué. Une file bornée transforme cela en erreur explicite que vous pouvez traiter. Dans paho-mqtt, une fois max_queued_messages_set(n) atteint, le MQTTMessageInfo renvoyé par publish() porte rc == MQTT_ERR_QUEUE_SIZE et rien n'est mis en file : journalisez-le et décidez s'il faut abandonner, échantillonner ou ralentir.

Côté réception, l'équivalent est la vitesse de traitement de l'abonné. Les callbacks s'exécutent sur le thread réseau ou l'event loop du client ; tant que votre on_message est occupé, le client ne lit pas le socket, n'envoie pas de PUBACK et ne répond pas au keep-alive. Confiez le travail à une file et rendez la main immédiatement :

import queue, threading

work = queue.Queue(maxsize=10000)

def on_message(client, userdata, msg):
    work.put_nowait((msg.topic, msg.payload))   # ne jamais bloquer ici

def worker():
    while True:
        topic, payload = work.get()
        save_to_database(topic, payload)        # travail lent hors du thread réseau

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

Si la file se remplit malgré tout, le consommateur est réellement trop lent : faites-le monter en charge avec des shared subscriptions ou consultez les consommateurs lents. Dans le client Go de Paho, SetOrderMatters(false) permet aux handlers de messages de s'exécuter en parallèle plutôt que l'un après l'autre, au prix des garanties d'ordre entre handlers.

5. Keep-alive et TCP_NODELAY

Le keep-alive n'accélère rien. Il détermine la vitesse à laquelle une connexion morte est détectée : le broker coupe un client après environ 1,5 × l'intervalle de keep-alive sans trafic. 30 à 60 secondes conviennent à la plupart des appareils connectés ; allez plus haut sur les liens sur batterie. La plupart des bibliothèques omettent le PINGREQ tant que d'autres paquets circulent. Détails dans MQTT keep alive.

L'algorithme de Nagle peut retarder les petits messages : le système retient un petit segment tant que des données précédentes ne sont pas acquittées et, combiné aux ACK retardés de l'autre côté, cela produit des pics de latence de plusieurs dizaines de millisecondes. Le désactiver avec TCP_NODELAY envoie chaque paquet immédiatement. Que votre bibliothèque le fasse déjà dépend de la bibliothèque et de sa version ; le positionner vous-même est sans danger. Dans paho-mqtt, faites-le dans le callback on_socket_open :

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

Mesurer avant et après

Optimiser sans mesurer, c'est deviner. Ce script publie un nombre fixe de messages QoS 1 avec une fenêtre inflight donnée et affiche le rythme auquel ils ont été acquittés :

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} acquittés en {secs:.2f} s = {acked / secs:.0f} msg/s")
client.loop_stop()
client.disconnect()

Lancez python inflight_bench.py 10, puis 20, 50, 100, 200. Le débit doit monter puis plafonner ; le coude de la courbe correspond à la bonne fenêtre pour ce chemin réseau. Répétez depuis l'endroit où se trouvent réellement vos appareils : un résultat mesuré à côté du broker en dit peu sur un appareil situé sur un autre continent. Pendant l'exécution, ouvrez le client en ligne et abonnez-vous à testmqtt/tuning/# pour confirmer que les messages arrivent.

Pour des tests plus importants avec de nombreux clients, passez à des outils dédiés — le guide des tests de débit les présente.

Optimisez dans cet ordre : cessez d'attendre chaque publication, augmentez la fenêtre inflight jusqu'à ce que la courbe plafonne, regroupez seulement si l'overhead par message domine encore, bornez vos files et gardez des callbacks rapides. Chaque étape tient en une ou deux lignes de code, et le script de benchmark vous dit si elle en valait la peine.

Questions fréquentes