Test de fan-in MQTT : des milliers d'appareils, un backend

Le fan-in est la forme quotidienne de l'IoT : des milliers d'appareils publient chacun un peu de télémétrie, et un backend s'abonne à devices/+/telemetry pour tout stocker ou traiter. Pris un par un, les appareils sont discrets ; ensemble, ils font tout passer par un abonnement, une connexion TCP et un processus. Un test de fan-in détermine combien d'appareils ce consommateur peut absorber avant qu'un backlog ne se forme — et comment le faire monter en charge quand il ne suit plus.

Ce guide simule une flotte avec emqtt-bench, mesure le backlog et le retard côté consommateur avec un script Python et montre comment les shared subscriptions répartissent la charge. Le schéma inverse est traité dans le test de fan-out.

Où se cachent les goulots d'étranglement du fan-in

Le broker gère généralement sans peine l'entrée du fan-in : router beaucoup de petits messages vers un subscriber coûte peu. Le problème se situe en aval :

Les deux dernières lignes sont des limites du broker sur une seule session — par exemple max_queued_messages et max_inflight_messages dans Mosquitto. Quand le consommateur prend du retard, le broker met des messages en file pour lui et, une fois la file pleine, il commence à supprimer ou à déconnecter selon le broker. Voir consommateurs lents pour le comportement des différents brokers.

Étape 1 : simuler la flotte

emqtt-bench peut jouer des milliers d'appareils depuis une seule machine. Donnez à chaque client son propre topic avec %i (l'index du client) pour que l'arborescence corresponde à la production. L'intervalle -I est par client et en millisecondes : 5 000 appareils avec -I 5000 envoient chacun un message toutes les 5 secondes, soit 1 000 msg/s au total :

docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 5000 -i 5 -I 5000 \
  -t 'testmqtt/fanin/%i/telemetry' -s 200 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

Pour augmenter la charge, baissez -I (5000 → 2000 → 1000 ms donne 2 500 puis 5 000 msg/s) plutôt que d'ajouter des connexions : vous testez le consommateur, pas le nombre de connexions. Avant de commencer, vérifiez avec le topic matcher que votre wildcard correspond bien aux topics des appareils — une faute de frappe dans le filtre ressemble exactement à un consommateur qui ne reçoit rien.

Étape 2 : un consommateur qui montre son backlog

Un consommateur de production ne doit jamais faire de travail lent dans on_message : cela bloque la boucle réseau, qui cesse de lire le socket et repousse le backlog dans le broker, où vous ne le voyez pas. Ce consommateur paho-mqtt 2.x confie chaque message à une file, le traite dans des threads de travail avec un coût simulé par message et affiche chaque seconde le débit reçu, le débit traité et le backlog :

import json, queue, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = "testmqtt/fanin/+/telemetry"
SHARED = "--shared" in sys.argv
TOPIC = f"$share/ingest/{FILTER}" if SHARED else FILTER
WORK_MS = 0.5        # coût simulé par message (parsing + insert en base)
WORKERS = 4

inbox = queue.Queue(maxsize=100_000)
counts = {"rx": 0, "done": 0}
lags = []

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    counts["rx"] += 1
    inbox.put(msg.payload)       # jamais de travail lent dans le thread réseau

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # messages sonde horodatés
            try:
                lags.append((time.time() - json.loads(payload)["t"]) * 1000)
            except (ValueError, KeyError):
                pass
        time.sleep(WORK_MS / 1000)
        counts["done"] += 1

for _ in range(WORKERS):
    threading.Thread(target=worker, daemon=True).start()

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                client_id=f"ingest-{uuid.uuid4().hex[:6]}", protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect(HOST, PORT)
c.loop_start()

while True:
    time.sleep(1)
    l = sorted(lags); lags.clear()
    p99 = f"{l[int(len(l) * 0.99) - 1]:.0f} ms" if l else "-"
    print(f"rx {counts['rx']:>6}/s  done {counts['done']:>6}/s  "
          f"backlog {inbox.qsize():>6}  probe lag p99 {p99}")
    counts.update(rx=0, done=0)

Les payloads d'emqtt-bench sont des octets aléatoires et ne peuvent donc pas porter d'horodatage. Pour mesurer le retard de bout en bout, lancez sur la même machine que le consommateur (pour partager la même horloge) une petite sonde qui publie du JSON horodaté dans la même arborescence de topics :

import json, time
import paho.mqtt.client as mqtt

p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
p.connect("broker.connectmq.com", 1883)
p.loop_start()
while True:
    p.publish("testmqtt/fanin/probe/telemetry", json.dumps({"t": time.time()}), qos=1)
    time.sleep(0.1)

Lire les chiffres

Faites tourner la flotte à chaque palier et observez la sortie du consommateur pendant quelques minutes :

• rx ≈ done, backlog proche de zéro, retard stable — le consommateur suit. Passez au palier suivant.
• rx ≈ entrée mais done < rx, backlog qui grossit — la partie réseau va bien, le traitement est le goulot. Ajoutez des workers, regroupez les écritures ou accélérez la destination.
• rx < entrée, backlog faible — la limite est la connexion unique ou la session côté broker (fenêtre inflight, file par client). C'est là qu'une shared subscription aide.
• Retard en hausse constante — vous êtes au-dessus du débit soutenable même si rien n'est encore perdu ; la file s'allonge simplement.
• Consommateur déconnecté sous charge — le broker a atteint sa limite par client, ou le keep-alive a expiré parce que le thread réseau était bloqué.

Étape 3 : monter en charge avec les shared subscriptions

Quand la limite est une seule connexion consommateur, lancez plusieurs consommateurs dans le même groupe de shared subscription. Le broker livre alors chaque message à un seul membre de $share/ingest/... :

python ingest.py --shared   # terminal 1
python ingest.py --shared   # terminal 2 (et ainsi de suite)

Rejouez le palier qui échouait. À vérifier :

1. La somme des done de tous les consommateurs correspond à l'entrée de la flotte — aucun message perdu.
2. Pas de doublons — sans $share, chaque consommateur recevrait tous les messages et vous traiteriez tout deux fois. Comptez les IDs uniques si votre payload en contient.
3. Répartition équilibrée — les brokers utilisent des stratégies différentes (round robin, aléatoire, sticky par client). Vérifiez que la charge est à peu près équilibrée.
4. Failover — tuez un consommateur pendant l'exécution ; les autres doivent absorber sa part. Les messages en vol vers le consommateur arrêté peuvent être relivrés ou perdus selon le QoS et le broker.
5. Ordre — l'ordre par appareil n'est pas garanti à l'échelle du groupe. S'il compte, concevez en conséquence.

Shared subscriptions explique la sémantique, et le test des shared subscriptions détaille les tests de répartition et de failover.

Au-delà d'un seul processus

Si la destination est le goulot et que les rafales sont inévitables — une flotte qui se reconnecte après une panne renvoie d'un coup beaucoup de données mises en tampon —, placez un tampon durable entre MQTT et le stockage. Faire un pont de MQTT vers Kafka est un schéma courant : voir MQTT vers Kafka. Testez ensuite explicitement le cas de rafale : arrêtez le consommateur cinq minutes pendant que la flotte publie, redémarrez-le et mesurez le temps nécessaire pour résorber le backlog. Pour un comportement d'appareil plus réaliste qu'avec emqtt-bench, utilisez un simulateur d'appareils avec vos vrais payloads.

Avant une grosse exécution, une vérification rapide dans le client MQTT en ligne — abonnez-vous à testmqtt/fanin/# et regardez les messages arriver — confirme que la flotte publie bien là où vous le pensez.

Le fan-in casse rarement le broker ; il casse le consommateur. Mesurez backlog et retard plutôt que des débits bruts, sortez le travail lent du thread réseau et montez en charge avec les shared subscriptions dès qu'une connexion ne suffit plus.

Questions fréquentes