Prueba de fan-in MQTT: miles de dispositivos, un backend

El fan-in es la forma cotidiana del IoT: miles de dispositivos publican cada uno un poco de telemetría y un backend se suscribe a devices/+/telemetry para almacenarla o procesarla toda. Por separado los dispositivos son discretos; juntos lo envían todo a través de una suscripción, una conexión TCP y un proceso. Una prueba de fan-in averigua cuántos dispositivos puede absorber ese consumidor antes de que se forme un backlog, y cómo escalarlo cuando no puede.

Esta guía simula una flota con emqtt-bench, mide el backlog y el retraso en el lado del consumidor con un script de Python y muestra cómo las shared subscriptions reparten la carga. El patrón inverso se trata en la prueba de fan-out.

Dónde se esconden los cuellos de botella del fan-in

El broker suele manejar sin problema la entrada del fan-in: enrutar muchos mensajes pequeños hacia un subscriber es barato. El problema está aguas abajo:

Los dos últimos son límites del broker sobre una sola sesión, por ejemplo max_queued_messages y max_inflight_messages en Mosquitto. Cuando el consumidor se retrasa, el broker le encola mensajes y, cuando la cola se llena, empieza a descartar o a desconectar, según el broker. Consulta consumidores lentos para ver cómo se comporta cada uno.

Paso 1: simular la flota

emqtt-bench puede hacer de miles de dispositivos desde una máquina. Da a cada cliente su propio topic con %i (el índice del cliente) para que la estructura de topics coincida con producción. El intervalo -I es por cliente y en milisegundos, así que 5.000 dispositivos con -I 5000 envían un mensaje cada 5 segundos cada uno: 1.000 msg/s en 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"

Para aumentar la carga, baja -I (5000 → 2000 → 1000 ms da 2.500 y 5.000 msg/s) en lugar de añadir conexiones: estás probando el consumidor, no el número de conexiones. Antes de empezar, comprueba con el topic matcher que tu wildcard coincide de verdad con los topics de los dispositivos: un error en el filtro se parece exactamente a un consumidor que no recibe nada.

Paso 2: un consumidor que muestra su backlog

Un consumidor de estilo producción nunca debería hacer trabajo lento dentro de on_message: eso bloquea el bucle de red, que deja de leer el socket y empuja el backlog al broker, donde no lo ves. Este consumidor con paho-mqtt 2.x pasa cada mensaje a una cola, lo procesa en hilos de trabajo con un coste simulado por mensaje e imprime cada segundo la tasa de recepción, la tasa de procesamiento y el 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        # coste simulado por mensaje (parseo + insert en BD)
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)       # nunca hagas trabajo lento en el hilo de red

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # mensajes de sonda con marca de tiempo
            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)

Los payloads de emqtt-bench son bytes aleatorios, así que no pueden llevar una marca de tiempo. Para medir el retraso extremo a extremo, ejecuta una pequeña sonda en la misma máquina que el consumidor (para que ambos usen el mismo reloj) que publique JSON con marca de tiempo en el mismo árbol 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)

Leer las cifras

Ejecuta la flota en cada paso y observa la salida del consumidor durante unos minutos:

• rx ≈ done, backlog cercano a cero, retraso plano: el consumidor da abasto. Pasa al siguiente paso.
• rx ≈ entrada pero done < rx, backlog creciendo: la parte de red va bien, el cuello de botella es el procesamiento. Añade workers, agrupa escrituras o acelera el destino.
• rx < entrada, backlog pequeño: el límite es la conexión única o la sesión en el broker (ventana inflight, cola por cliente). Aquí ayuda una shared subscription.
• Retraso subiendo sin parar: estás por encima de la tasa sostenible aunque todavía no se pierda nada; la cola simplemente se alarga.
• Consumidor desconectado bajo carga: el broker alcanzó su límite por cliente o expiró el keep-alive porque el hilo de red estaba bloqueado.

Paso 3: escalar con shared subscriptions

Cuando el límite es una sola conexión de consumidor, arranca varios consumidores en el mismo grupo de shared subscription. El broker entrega entonces cada mensaje a un solo miembro de $share/ingest/...:

python ingest.py --shared   # terminal 1
python ingest.py --shared   # terminal 2 (y así sucesivamente)

Repite el paso que antes fallaba. Qué verificar:

1. La suma de done de todos los consumidores coincide con la entrada de la flota: no se pierden mensajes.
2. Sin duplicados: sin $share, cada consumidor recibiría todos los mensajes y lo procesarías todo dos veces. Cuenta IDs únicos si tu payload los lleva.
3. Reparto equilibrado: los brokers usan estrategias distintas (round robin, aleatoria, sticky por cliente). Comprueba que la carga está más o menos equilibrada.
4. Failover: mata un consumidor durante la ejecución; los demás deben absorber su parte. Los mensajes en vuelo hacia el consumidor caído pueden reenviarse o perderse según el QoS y el broker.
5. Orden: el orden por dispositivo no está garantizado a lo largo del grupo. Si importa, diseña para ello.

Shared subscriptions explica la semántica, y la prueba de shared subscriptions cubre con más detalle las pruebas de reparto y failover.

Más allá de un proceso

Si el cuello de botella es el destino y las ráfagas son inevitables —una flota que se reconecta tras una caída reenvía de golpe muchos datos almacenados—, pon un buffer duradero entre MQTT y el almacenamiento. Puentear MQTT hacia Kafka es un patrón habitual: consulta MQTT a Kafka. Después prueba explícitamente el caso de ráfaga: detén el consumidor cinco minutos mientras la flota sigue publicando, arráncalo de nuevo y mide cuánto tarda en vaciarse el backlog. Para un comportamiento de dispositivo más realista que el de emqtt-bench, usa un simulador de dispositivos con tus payloads reales.

Antes de una ejecución grande, una comprobación rápida en el cliente MQTT online —suscríbete a testmqtt/fanin/# y mira cómo llegan los mensajes— confirma que la flota publica donde crees.

El fan-in rara vez rompe el broker; rompe el consumidor. Mide backlog y retraso en lugar de tasas de recepción en bruto, saca el trabajo lento del hilo de red y escala con shared subscriptions cuando una conexión no baste.

Preguntas frecuentes