Outils de test de charge MQTT : emqtt-bench, MQTTX Bench et Locust comparés

Les outils de test de charge MQTT vous indiquent si votre système tient le coup lorsque des milliers d'appareils se connectent en même temps, qu'un bug de firmware inonde un topic ou qu'un subscriber backend prend du retard. Ce guide compare emqtt-bench, MQTTX Bench et Locust, présente les commandes de chacun et ajoute une petite sonde Python qui mesure la latence de bout en bout et la perte de messages. Vous verrez aussi quelles métriques comptent et comment tester de façon responsable, toujours sur votre propre broker privé, jamais sur un broker public partagé.

Règle n° 1 : testez votre propre broker

Choisir parmi les outils de test de charge MQTT disponibles est la partie facile ; les pointer vers la bonne cible est ce qui compte. Les tests de charge génèrent un trafic qui ressemble à une attaque. Ne les exécutez que sur une infrastructure qui vous appartient ou que vous êtes explicitement autorisé à tester.

• Ne faites pas de test de charge sur le serveur public. broker.connectmq.com, le serveur MQTT public gratuit, est un service de test partagé avec des limites strictes par client (10 messages par seconde par client, 10 abonnements par client, messages de 64 Ko). Un test de charge y mesure ces limites, pas votre système, et dégrade le service pour tous les autres.
• Utilisez un broker privé dimensionné pour le test. Sur ConnectMQ, créez un broker dédié aux tests de charge pour que les appareils de production ne soient pas affectés.
• Connaissez les quotas de votre plan avant de commencer. Le plan Serverless gratuit autorise jusqu'à 25 connexions et une QoS jusqu'à 1 ; le Serverless payant autorise jusqu'à 1 000 connexions, la QoS 2 et des messages de 1 Mo ; Dedicated tourne sur un cluster isolé. Dépasser un quota n'est pas une panne du broker : les clients MQTT 5 reçoivent le reason code 0x97 (quota exceeded), que votre test doit enregistrer séparément des vraies erreurs.
• Vérifiez les règles de client ID. ConnectMQ lie les client ID aux identifiants qu'il émet, alors que les outils de charge génèrent des ID à partir d'un préfixe et d'un compteur. Confirmez dans le portail quels client ID vos identifiants de test autorisent et configurez l'option de client ID de l'outil en conséquence, sinon chaque connexion sera rejetée comme non autorisée.
• Tenez compte des couches de protection. Les brokers MQTT managés appliquent une protection contre le flood et bloquent les IP après des échecs de connexion répétés. Un test mal configuré avec de mauvais identifiants peut faire bloquer l'IP de votre machine de test : validez donc d'abord les identifiants avec un seul client.

Ce qu'il faut mesurer

Le débit seul est un chiffre trompeur. Collectez les métriques suivantes, et fixez les critères de réussite avant de lancer quoi que ce soit :

Surveillez toujours le générateur de charge. Si son CPU est à 100 %, les chiffres de latence reflètent votre portable, pas le broker. Répartissez les gros tests sur plusieurs machines.

emqtt-bench

emqtt-bench est un outil Erlang léger issu du projet EMQX, conçu pour ouvrir de nombreuses connexions avec très peu de mémoire par client. Il propose trois modes : conn, pub et sub.

# Ramp up 500 connections, one every 10 ms
emqtt_bench conn -h YOUR_BROKER_HOST -p 1883 -c 500 -i 10 \
  -u YOUR_USERNAME -P YOUR_PASSWORD

# 50 subscribers on a wildcard, QoS 1
emqtt_bench sub -h YOUR_BROKER_HOST -p 1883 -c 50 -i 10 \
  -t 'loadtest/#' -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

# 200 publishers, each sending a 256-byte message every 1000 ms to its own topic
emqtt_bench pub -h YOUR_BROKER_HOST -p 1883 -c 200 -i 10 -I 1000 \
  -t 'loadtest/device/%i' -s 256 -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

Flags clés : -c nombre de clients, -i intervalle entre deux nouvelles connexions en millisecondes, -I intervalle entre publications par client, -t topic avec les placeholders %i (numéro de séquence du client) et %c (client ID), -s taille du payload, -q QoS. Ajoutez --ssl et le port 8883 pour tester TLS, ce que vous devriez faire, car les négociations TLS sont souvent le vrai coût lors des tempêtes de reconnexion. L'outil affiche les débits toutes les quelques secondes.

Points forts : très efficace par connexion, idéal pour les tests de nombre de connexions et de tempêtes de reconnexion. Points faibles : les payloads sont synthétiques et la mesure de latence est limitée ; associez-le donc à une sonde (voir plus bas).

MQTTX Bench

MQTTX CLI inclut une commande bench avec les trois mêmes modes et une sortie plus conviviale. Si votre équipe utilise déjà MQTTX, c'est le point de départ le plus simple.

mqttx bench conn -c 200 -i 10 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench sub -c 20 -t 'loadtest/#' -q 1 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench pub -c 100 -i 10 -im 1000 -t 'loadtest/device/%i' -q 1 \
  -m '{"temp": 21.5}' -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

Ici, -i est l'intervalle de connexion et -im l'intervalle entre messages par client. Comme il tourne sur Node.js, chaque client consomme plus de mémoire qu'avec emqtt-bench : les très grands nombres de connexions nécessitent donc davantage de machines génératrices. Consultez mqttx bench pub --help dans votre version, car les options évoluent.

Locust

Locust est un framework Python de test de charge. Il n'a pas de support MQTT natif, mais vous pouvez écrire une classe User autour de paho-mqtt et remonter les résultats dans l'interface web de Locust. C'est l'option la plus flexible : payloads réalistes, comportement par appareil, profils de montée en charge et assertions ne sont que du Python.

# locustfile.py  (pip install locust paho-mqtt)
import json
import os
import random
import time
import uuid

import paho.mqtt.client as mqtt
from locust import User, between, events, task

HOST = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST")
USER = os.environ.get("MQTT_USER", "YOUR_USERNAME")
PASS = os.environ.get("MQTT_PASS", "YOUR_PASSWORD")


class MqttDevice(User):
    wait_time = between(1, 2)

    def on_start(self):
        self.device_id = uuid.uuid4().hex[:8]
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"lt-{self.device_id}",
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(USER, PASS)
        self.client.tls_set()
        start = time.perf_counter()
        try:
            self.client.connect(HOST, 8883, keepalive=30)
            self.client.loop_start()
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=None, context={})
        except Exception as exc:
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=exc, context={})
            raise

    @task
    def publish_telemetry(self):
        payload = json.dumps({"id": self.device_id,
                              "temp": round(random.uniform(18, 26), 2),
                              "ts": time.time()})
        start = time.perf_counter()
        info = self.client.publish(f"loadtest/device/{self.device_id}", payload, qos=1)
        exc = None
        try:
            info.wait_for_publish(timeout=5)  # waits for PUBACK at QoS 1
            if not info.is_published():
                exc = TimeoutError("no PUBACK within 5 s")
        except Exception as e:
            exc = e
        events.request.fire(request_type="MQTT", name="publish qos1",
                            response_time=(time.perf_counter() - start) * 1000,
                            response_length=len(payload), exception=exc, context={})

    def on_stop(self):
        self.client.loop_stop()
        self.client.disconnect()

Lancez-le avec locust -f locustfile.py --headless -u 100 -r 5 --run-time 10m, où -u est le nombre d'appareils simulés et -r le taux d'apparition par seconde. Chaque appareil exécute son propre thread réseau paho : Locust est donc plus lourd par connexion qu'emqtt-bench ; utilisez-le pour des tests riches en comportements avec des centaines d'appareils plutôt que pour des nombres bruts de connexions. Le chronométrage de publication mesure ici l'aller-retour du PUBACK, un bon indicateur de la réactivité du broker.

Une sonde Python de latence et de perte

Aucun des outils ci-dessus ne fournit par défaut une mesure fiable de la latence de bout en bout et de la perte. Exécutez cette sonde en parallèle de n'importe quel test de charge : elle publie des messages séquencés et horodatés, et mesure ce qui arrive.

# probe.py  (pip install paho-mqtt)
import json, os, statistics, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST"), 8883
USER, PASS = os.environ.get("MQTT_USER", "YOUR_USERNAME"), os.environ.get("MQTT_PASS", "YOUR_PASSWORD")
TOPIC = f"loadtest/probe/{uuid.uuid4().hex[:6]}"
COUNT, INTERVAL = 600, 0.5

latencies, seen = [], set()

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    latencies.append((time.time() - data["ts"]) * 1000)
    seen.add(data["seq"])

def make(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{name}-{uuid.uuid4().hex[:6]}")
    c.username_pw_set(USER, PASS)
    c.tls_set()
    c.connect(HOST, PORT, keepalive=30)
    c.loop_start()
    return c

sub = make("sub")
sub.on_message = on_message
sub.subscribe(TOPIC, qos=1)
time.sleep(1)
pub = make("pub")

for seq in range(COUNT):
    pub.publish(TOPIC, json.dumps({"seq": seq, "ts": time.time()}), qos=1)
    time.sleep(INTERVAL)

time.sleep(5)
q = statistics.quantiles(latencies, n=100)
print(f"received {len(seen)}/{COUNT}, lost {COUNT - len(seen)}")
print(f"latency ms p50={q[49]:.1f} p95={q[94]:.1f} p99={q[98]:.1f} max={max(latencies):.1f}")

Le publisher et le subscriber tournent sur la même machine : le décalage d'horloge ne fausse donc pas la latence. Exécutez la sonde une fois sur un broker au repos pour obtenir une référence, puis pendant chaque palier de charge ; la différence correspond au coût de la charge. Si vos identifiants n'autorisent que des client ID précis, remplacez les ID générés par ceux qui vous ont été attribués.

Synthèse comparative

Un plan de test responsable

1. Formulez l'objectif : par exemple, « 2 000 appareils se reconnectent en moins de 2 minutes ; la latence p95 reste sous un seuil convenu ».
2. Validez avec un seul client : identifiants, TLS, structure des topics, règles de client ID.
3. Établissez une référence : exécutez la sonde sur un broker au repos.
4. Montez par paliers : 10, 25, 50 puis 100 % de la cible, en maintenant chaque palier plusieurs minutes.
5. Testez les cas difficiles : tout le monde se reconnecte en même temps, un subscriber cesse de lire, des payloads surdimensionnés.
6. Enregistrez séparément les réponses de quota : 0x97 signifie que vous avez atteint une limite du plan, c'est une information, pas un plantage.
7. Faites le ménage : supprimez les retained messages de test et déconnectez tous les clients.

Si les résultats montrent qu'il vous faut plus de connexions ou d'isolation que votre plan actuel n'en offre, c'est le signal pour monter en gamme, par exemple de Serverless à Dedicated. La présentation du broker MQTT managé et le guide pour choisir un broker MQTT cloud expliquent les compromis. Pour les tests fonctionnels avant les tests de charge, consultez tester les clients MQTT, et pour générer un trafic réaliste à faible volume, utilisez un simulateur d'appareils MQTT. Les bases de la ligne de commande sont traitées dans les outils CLI MQTT, et le guide des bibliothèques client MQTT vous aide si vous voulez créer un générateur de charge personnalisé dans un autre langage.

Démarrer avec ConnectMQ

Créez un broker de test dédié sur app.connectmq.com avec le plan Serverless gratuit, sans carte bancaire, et validez vos scripts de test à petite échelle. Lorsque vous êtes prêt pour des campagnes plus importantes, faites évoluer la même configuration de broker et gardez le tableau de bord du portail ouvert pour suivre les connexions pendant le test.

Questions fréquentes

Peut-on faire un test de charge sur le serveur MQTT public gratuit ?

Non. Les brokers publics sont partagés et limités en débit par client : les résultats n'ont aucune valeur et les autres utilisateurs en pâtissent. Utilisez votre propre broker privé.

Quel est le meilleur outil de test de charge MQTT ?

emqtt-bench pour les grands nombres de connexions, MQTTX Bench pour des tests rapides et lisibles, et Locust lorsque vous avez besoin d'un comportement d'appareil réaliste. Beaucoup d'équipes combinent emqtt-bench pour la montée en charge et une petite sonde pour la latence.

Pourquoi mes clients de test de charge sont-ils rejetés avec quota exceeded ?

Vous avez atteint une limite de votre plan, comme le nombre maximal de connexions. Les clients MQTT 5 reçoivent le reason code 0x97 ; réduisez la charge ou passez à un plan aux limites plus élevées.

Les tests de charge doivent-ils utiliser TLS ?

Oui, si la production utilise TLS. Les négociations TLS ajoutent une charge CPU importante des deux côtés lors des tempêtes de reconnexion, et tester sans TLS masque ce coût.