Prueba de fan-out MQTT: un publisher, miles de subscribers

En un escenario de fan-out un mensaje llega a muchos receptores: un cambio de configuración enviado a toda una flota, una actualización de precio mostrada en 10.000 dashboards, un aviso de "firmware disponible". El publisher envía un PUBLISH; el broker envía uno por cada suscripción coincidente. Una prueba de fan-out mide cuánto tarda el último subscriber en recibir el mensaje y si todos lo recibieron.

Esta guía monta un benchmark 1 → 10.000, añade una sonda que mide la dispersión de entrega y hace el cálculo de ancho de banda para que la factura de red no te sorprenda. Para pruebas generales de tasa, consulta la prueba de throughput MQTT; para el patrón inverso, la prueba de fan-in.

Por qué el fan-out es una prueba aparte

Las pruebas de throughput de publicación suelen emparejar cada publisher con un subscriber. El fan-out estresa algo distinto: la capacidad del broker de copiar un mensaje en miles de colas de salida y vaciarlas hacia miles de sockets.

La pregunta no es "cuántos mensajes por segundo", sino "cuánto tarda hasta que todos los subscribers lo tienen" —la dispersión de entrega— con el tamaño real de tu flota.

Primero, el cálculo de ancho de banda

Haz las cuentas antes de empezar: una prueba de fan-out puede saturar un enlace de red mucho antes de que el broker sufra. Un PUBLISH en la red es aproximadamente cabecera fija + 2 bytes de longitud del topic + topic + identificador de paquete (QoS > 0) + propiedades + payload. Consulta la sobrecarga de ancho de banda de MQTT para los detalles.

Ejemplo: payload de 500 bytes en testmqtt/fanout/broadcast (25 bytes) con QoS 0 ≈ 530 bytes por PUBLISH.

Suma encima la sobrecarga de TCP/IP y TLS. La última fila supera lo que muchas máquinas generadoras de carga pueden recibir solas, por eso las pruebas de fan-out suelen repartirse en varias máquinas. La calculadora de tamaño de paquete te da el tamaño exacto por paquete para tu topic y payload.

Paso 1: abrir los subscribers

emqtt-bench puede mantener miles de conexiones de subscriber. Reparte 10.000 entre cuatro máquinas (o contenedores en hosts distintos) para no chocar con los límites de descriptores de archivo y de puertos efímeros:

ulimit -n 65535
docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench sub \
  -h broker.connectmq.com -p 1883 \
  -c 2500 -i 5 \
  -t testmqtt/fanout/broadcast -q 0 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

-i 5 abre una conexión cada 5 ms (200 por segundo y máquina), así que las cuatro máquinas necesitan algo más de 12 segundos para conectar a todos. Abrir conexiones más rápido es una prueba en sí misma —consulta la prueba de estrés de conexiones— y no debe mezclarse con las mediciones de fan-out. Espera a que las cuatro informen del total antes de publicar.

Paso 2: publicar y sondear

La salida de subscriber de emqtt-bench te da la tasa agregada de recepción, pero no cuánto se retrasó el subscriber más lento. Este script con paho-mqtt 2.x añade 50 subscribers sonda en un proceso, publica mensajes con marca de tiempo en el mismo topic de broadcast e informa de la latencia y la dispersión de entrega entre las sondas. Como el publisher y las sondas comparten reloj, las cifras son directamente comparables:

import json, threading, time, uuid
from collections import defaultdict
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = "testmqtt/fanout/broadcast"
PROBES, MESSAGES, RATE, QOS = 50, 60, 1, 0
lat = defaultdict(list)          # seq -> latencias (ms) entre sondas
lock = threading.Lock()

def on_message(client, userdata, msg):
    try:
        m = json.loads(msg.payload)
    except ValueError:
        return                   # ignorar tráfico que no es de la sonda
    with lock:
        lat[m["seq"]].append((time.perf_counter() - m["t"]) * 1000)

def new_client(name):
    return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                       client_id=f"{name}-{uuid.uuid4().hex[:6]}")

probes = []
for n in range(PROBES):
    c = new_client(f"fanout-probe-{n}")
    c.on_message = on_message
    c.connect(HOST, PORT)
    c.subscribe(TOPIC, qos=QOS)
    c.loop_start()
    probes.append(c)

pub = new_client("fanout-pub")
pub.connect(HOST, PORT)
pub.loop_start()
time.sleep(3)                    # dejar que se asienten las suscripciones

for seq in range(MESSAGES):
    body = {"seq": seq, "t": time.perf_counter(), "pad": "x" * 450}
    pub.publish(TOPIC, json.dumps(body), qos=QOS)
    time.sleep(1 / RATE)
time.sleep(5)

pct = lambda xs, q: xs[min(len(xs) - 1, int(len(xs) * q))]
all_lat = sorted(v for vals in lat.values() for v in vals)
spread = sorted(max(v) - min(v) for v in lat.values() if len(v) > 1)
print(f"delivered {len(all_lat)}/{PROBES * MESSAGES}")
if all_lat:
    print(f"latency p50 {pct(all_lat, .5):.1f}  p99 {pct(all_lat, .99):.1f}  max {all_lat[-1]:.1f} ms")
if spread:
    print(f"spread  p50 {pct(spread, .5):.1f}  max {spread[-1]:.1f} ms")

Ejecútalo tres veces: sin subscribers del bench (línea base), con 2.500 y con los 10.000. La línea base te da la aportación de la red; el crecimiento a partir de ahí es el coste de fan-out del broker. Después repite con QOS = 1 y con un RATE mayor para ver dónde empieza a subir la latencia.

Interpretar los resultados

• Entregados < esperados en las sondas con QoS 0 significa que el broker descartó mensajes para algunas conexiones, normalmente porque sus colas de salida estaban llenas. Repite con QoS 1 y revisa los logs del broker.
• La latencia p99 crece con el número de subscribers mientras p50 se mantiene es normal hasta cierto punto: el broker escribe en los sockets uno tras otro. Lo que importa es si p99 sigue dentro de tu requisito.
• La dispersión se dispara con tasas de publicación más altas significa que el broker (o su red) no puede vaciar todas las copias antes de que llegue el siguiente mensaje. Has superado la tasa de fan-out sostenible.
• Las máquinas del bench informan de tasas de recepción menores de lo esperado: revisa primero su CPU y su red. Recibir 100.000 msg/s es trabajo real para un generador de carga.
• Algunos subscribers se quedan atrás y se desconectan: consulta consumidores lentos. En producción, un dispositivo lento no debe retrasar a todos los demás.

Diseñar topics para el fan-out

La prueba suele terminar en una conversación de diseño. Algunos patrones funcionan bien:

1. Broadcast solo donde todos lo necesitan de verdad. fleet/all/config para mensajes que todos los dispositivos deben recibir; mantenlos pequeños.
2. Topics de grupo en lugar de un broadcast gigante. fleet/region/eu/config o fleet/model/x200/firmware reducen el fan-out por mensaje y permiten despliegues graduales.
3. Retén el estado actual. Para configuración, publica con el flag retain para que los dispositivos que se reconecten más tarde reciban el último valor al instante; consulta mensajes retenidos. Prueba que un subscriber tardío lo recibe.
4. Elige el QoS según la consecuencia. Actualizaciones reemplazables con QoS 0; comandos que deben llegar con QoS 1, dimensionados para que el seguimiento por subscriber del broker siga siendo asumible.
5. Evita aquí las shared subscriptions. Balancean carga; no hacen broadcast.

Una lista de comprobación corta

1. Calcula el ancho de banda de salida para tu flota objetivo y tu tasa de publicación.
2. Reparte los subscribers en suficientes máquinas de carga; sube los límites de descriptores de archivo.
3. Conecta primero todos los subscribers y después publica; mantén separadas las pruebas de conexión.
4. Mide latencia y dispersión con sondas, no solo con tasas agregadas de recepción.
5. Repite con QoS 0 y QoS 1, con y sin TLS.
6. Registra CPU, memoria y red de salida del broker junto a las cifras del cliente.

El fan-out convierte un mensaje pequeño en una gran ráfaga de trabajo y tráfico. Haz primero el cálculo de ancho de banda, conecta los subscribers antes de publicar y juzga el resultado por el subscriber más lento: es el que notarán tus usuarios.

Preguntas frecuentes