Herramientas de pruebas de carga MQTT: comparativa de emqtt-bench, MQTTX Bench y Locust

Las herramientas de pruebas de carga MQTT te dicen si tu sistema aguanta cuando miles de dispositivos se conectan a la vez, un error de firmware inunda un topic o un suscriptor del backend se queda atrás. Esta guía compara emqtt-bench, MQTTX Bench y Locust, muestra los comandos de cada una y añade una pequeña sonda en Python que mide la latencia de extremo a extremo y la pérdida de mensajes. También aprenderás qué métricas importan y cómo probar de forma responsable, siempre contra tu propio broker privado y nunca contra uno público compartido.

Primera regla: prueba tu propio broker

Elegir entre las herramientas de pruebas de carga MQTT disponibles es la parte fácil; lo importante es apuntarlas al objetivo correcto. Las pruebas de carga generan tráfico que parece un ataque. Ejecútalas solo contra infraestructura que sea tuya o que tengas autorización expresa para probar.

• No hagas pruebas de carga contra el broker público. broker.connectmq.com es el servidor MQTT público gratuito, un servicio de pruebas compartido con límites estrictos por cliente (10 mensajes por segundo por cliente, 10 suscripciones por cliente, mensajes de 64 KB). Una prueba de carga ahí mide esos límites, no tu sistema, y lo degrada para todos los demás.
• Usa un broker privado dimensionado para la prueba. En ConnectMQ, crea un broker dedicado a las pruebas de carga para no afectar a los dispositivos de producción.
• Conoce las cuotas de tu plan antes de empezar. El plan Serverless gratuito permite hasta 25 conexiones y QoS hasta 1; Serverless de pago permite hasta 1.000 conexiones, QoS 2 y mensajes de 1 MB; Dedicated se ejecuta en un clúster aislado. Superar una cuota no es un fallo del broker: los clientes MQTT 5 reciben el reason code 0x97 (quota exceeded), que tu prueba debe registrar por separado de los errores reales.
• Revisa las reglas de los client ID. ConnectMQ vincula los client ID a las credenciales que emite, mientras que las herramientas de carga generan los ID a partir de un prefijo y un contador. Confirma en el portal qué client ID permiten tus credenciales de prueba y configura en consecuencia la opción de client ID de la herramienta; de lo contrario, todas las conexiones se rechazarán por no estar autorizadas.
• Ten en cuenta las capas de protección. Los brokers MQTT gestionados aplican protección contra inundaciones y bloquean las IP tras varios inicios de sesión fallidos. Una prueba mal configurada con credenciales incorrectas puede hacer que se bloquee la IP de tu máquina de pruebas, así que valida primero las credenciales con un solo cliente.

Qué medir

El caudal por sí solo es una cifra engañosa. Recoge estas métricas y decide los criterios de aceptación antes de ejecutar nada:

Vigila siempre el generador de carga. Si su CPU está al 100 por ciento, las cifras de latencia reflejan tu portátil, no el broker. Reparte las pruebas grandes entre varias máquinas.

emqtt-bench

emqtt-bench es una herramienta ligera en Erlang del proyecto EMQX, diseñada para abrir muchas conexiones con muy poca memoria por cliente. Tiene tres modos: conn, pub y sub.

# Rampa de 500 conexiones, una cada 10 ms
emqtt_bench conn -h YOUR_BROKER_HOST -p 1883 -c 500 -i 10 \
  -u YOUR_USERNAME -P YOUR_PASSWORD

# 50 suscriptores sobre un comodín, 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 publicadores, cada uno envía un mensaje de 256 bytes cada 1000 ms a su propio 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 clave: -c número de clientes, -i intervalo entre conexiones nuevas en milisegundos, -I intervalo entre publicaciones por cliente, -t topic con los marcadores %i (número de secuencia del cliente) y %c (client ID), -s tamaño del payload, -q QoS. Añade --ssl y el puerto 8883 para probar TLS, cosa que deberías hacer, porque los handshakes TLS suelen ser el coste real durante las avalanchas de reconexión. La herramienta imprime las tasas cada pocos segundos.

Puntos fuertes: muy eficiente por conexión, ideal para pruebas de número de conexiones y de avalanchas de reconexión. Puntos débiles: los payloads son sintéticos y la medición de latencia es limitada, así que combínala con una sonda (más abajo).

MQTTX Bench

MQTTX CLI incluye un comando bench con los mismos tres modos y una salida más amigable. Si tu equipo ya usa MQTTX, es el punto de partida más sencillo.

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

Aquí -i es el intervalo de conexión e -im el intervalo de mensajes por cliente. Como se ejecuta sobre Node.js, cada cliente consume más memoria que con emqtt-bench, así que los números de conexiones muy grandes necesitan más máquinas generadoras. Consulta mqttx bench pub --help en tu versión, porque las opciones cambian.

Locust

Locust es un framework de pruebas de carga en Python. No tiene soporte nativo para MQTT, pero puedes escribir una clase User en torno a paho-mqtt y enviar los resultados a la interfaz web de Locust. Eso la convierte en la opción más flexible: payloads realistas, comportamiento por dispositivo, formas de rampa y aserciones son simplemente 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)  # espera el PUBACK con 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()

Ejecútalo con locust -f locustfile.py --headless -u 100 -r 5 --run-time 10m, donde -u es el número de dispositivos simulados y -r la tasa de creación por segundo. Cada dispositivo ejecuta su propio hilo de red de paho, así que Locust es más pesado por conexión que emqtt-bench; úsalo para pruebas con comportamiento rico y cientos de dispositivos, más que para números brutos de conexiones. El tiempo de publicación medido aquí es el viaje de ida y vuelta hasta el PUBACK, un buen indicador de la capacidad de respuesta del broker.

Una sonda de latencia y pérdida en Python

Ninguna de las herramientas anteriores te da por defecto una latencia de extremo a extremo y una pérdida fiables. Ejecuta esta sonda junto a cualquier prueba de carga: publica mensajes numerados y con marca de tiempo y mide lo que llega.

# 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}")

El publicador y el suscriptor se ejecutan en la misma máquina, así que el desfase de reloj no distorsiona la latencia. Ejecuta la sonda una vez con el broker en reposo para obtener una referencia y luego durante cada fase de carga; la diferencia es lo que cuesta la carga. Si tus credenciales solo permiten client ID concretos, sustituye los ID generados por los que te emitieron.

Resumen comparativo

Un plan de pruebas responsable

1. Escribe el objetivo: por ejemplo, «2.000 dispositivos se reconectan en 2 minutos; la latencia p95 se mantiene por debajo de un umbral acordado».
2. Valida con un solo cliente: credenciales, TLS, estructura de topics, reglas de client ID.
3. Referencia: ejecuta la sonda con el broker en reposo.
4. Aumenta por fases: 10, 25, 50 y 100 por ciento del objetivo, manteniendo cada fase varios minutos.
5. Prueba los casos difíciles: todos reconectándose a la vez, un suscriptor que deja de leer, payloads sobredimensionados.
6. Registra por separado las respuestas de cuota: 0x97 significa que alcanzaste un límite del plan, lo cual es información, no un fallo.
7. Limpia: borra los mensajes retenidos de prueba y desconecta todos los clientes.

Si los resultados muestran que necesitas más conexiones o más aislamiento de lo que ofrece tu plan actual, esa es la señal para subir de nivel, por ejemplo de Serverless a Dedicated. La visión general del broker MQTT gestionado y la guía para elegir un broker MQTT en la nube explican las ventajas e inconvenientes. Para las pruebas funcionales previas a las de carga, consulta cómo probar clientes MQTT, y para generar tráfico realista a bajo volumen usa un simulador de dispositivos MQTT. Los fundamentos de línea de comandos se tratan en herramientas CLI de MQTT, y la guía de bibliotecas cliente MQTT te ayuda si quieres construir un generador de carga propio en otro lenguaje.

Empieza con ConnectMQ

Crea un broker de pruebas dedicado en app.connectmq.com con el plan Serverless gratuito, sin tarjeta de crédito, y valida tus scripts de prueba a pequeña escala. Cuando estés listo para ejecuciones más grandes, mejora el plan con la misma configuración de broker y mantén abierto el panel del portal para observar las conexiones mientras se ejecuta la prueba.

Preguntas frecuentes

¿Puedo hacer pruebas de carga contra el broker MQTT público?

No. Los brokers públicos son compartidos y tienen límites de tasa por cliente, así que los resultados no significan nada y perjudicas a otros usuarios. Usa tu propio broker privado.

¿Cuál es la mejor herramienta de pruebas de carga MQTT?

emqtt-bench para grandes números de conexiones, MQTTX Bench para pruebas rápidas y legibles, y Locust cuando necesitas un comportamiento realista de los dispositivos. Muchos equipos usan emqtt-bench para la escala más una pequeña sonda para la latencia.

¿Por qué mis clientes de prueba de carga se rechazan con quota exceeded?

Alcanzaste un límite de tu plan, como el número máximo de conexiones. Los clientes MQTT 5 reciben el reason code 0x97; reduce la carga o pasa a un plan con límites más altos.

¿Las pruebas de carga deben usar TLS?

Sí, si producción usa TLS. Los handshakes añaden un trabajo de CPU considerable en ambos extremos durante las avalanchas de reconexión, y probar sin TLS oculta ese coste.