Cómo probar las suscripciones compartidas y el balanceo de carga en MQTT

Una suscripción compartida ($share/<grupo>/<topic>) convierte varios suscriptores en un pool de workers: cada mensaje del topic se entrega a un único miembro del grupo en lugar de a todos. Para probarla, arranca varios workers en el mismo grupo, publica una secuencia numerada y comprueba tres cosas: cada mensaje llegó exactamente una vez al grupo, la carga se repartió y no se perdió nada cuando un worker cayó en mitad del flujo.

Esta guía muestra una comprobación rápida con la CLI, una prueba en Python con simulación de caída y los casos límite que merece la pena verificar. Los conceptos están en las suscripciones compartidas de MQTT. Aquí solo los probamos.

Qué hace un broker correcto

La estrategia de reparto la decide el broker. Mosquitto usa round robin. Otros brokers ofrecen estrategias aleatorias, sticky o basadas en hash, a menudo configurables. Tu prueba debe comprobar un reparto justo dentro de una tolerancia, no un patrón exacto, salvo que dependas de una estrategia concreta.

Paso 0: comprobar que el broker lo soporta

Un broker MQTT 5 anuncia el soporte en el CONNACK con la propiedad Shared Subscription Available. Si la propiedad no aparece, se asume que hay soporte. Si vale 0, el broker rechazará o ignorará las suscripciones $share, y el SUBACK devuelve 0x9E Shared Subscriptions not supported. En cualquier caso, la prueba real es la de reparto que verás más abajo. La consulta de reason codes descifra el SUBACK que obtengas.

El nombre del grupo es un único nivel de topic: sin /, + ni #. El resto es un filtro de topic normal, así que los wildcards funcionan después del grupo: $share/workers/testmqtt/share/+/jobs. Comprueba los filtros con el topic matcher antes de culpar al broker.

Comprobación rápida con la CLI de Mosquitto

Abre dos terminales con miembros del mismo grupo y una tercera con un suscriptor normal:

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t '$share/demo/testmqtt/share/9b1e/jobs' -v

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
    -t testmqtt/share/9b1e/jobs -m "job $i"
done

Las terminales 1 y 2 se reparten los diez trabajos (con round robin, los impares en un lado y los pares en el otro). La terminal 3 ve los diez. Pon siempre el topic entre comillas simples: $share sin comillas o entre comillas dobles lo expandiría tu shell como variable.

Prueba automatizada de reparto y failover

Este script de paho-mqtt 2.x arranca tres workers en un grupo, publica 100 mensajes QoS 1 numerados e informa del recuento por worker, los huecos y los duplicados. Con --failover hace caer a worker-0 a mitad de camino: deja de leer, así que sus últimos mensajes quedan sin confirmar, y luego corta la conexión TCP sin DISCONNECT, como haría una caída real.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

Una ejecución contra un broker con round robin se ve así:

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

Cómo leerlo:

• Sin --failover, los recuentos deberían ser parecidos (33/33/34 con round robin). Un reparto muy desequilibrado apunta a una estrategia sticky o por hash, o a un worker lento que el broker se salta.
• Con --failover, missing 0 significa que el broker redirigió el trabajo del miembro caído. Si faltan IDs justo alrededor del momento de la caída, los mensajes en vuelo se perdieron junto con la sesión. Algunos brokers esperan primero a que el cliente reconecte, así que prueba con tu configuración de sesión real.
• Los duplicados son legales en QoS 1: un mensaje puede entregarse al worker caído y reenviarse a otro. Los workers de un grupo compartido deben ser idempotentes. Consulta mensajes MQTT duplicados.
• Mensajes perdidos sin ninguna caída suelen indicar un límite de tasa o de cola en el broker. Baja RATE y vuelve a ejecutar. Los endpoints públicos compartidos son especialmente propensos a limitar.

Casos límite que conviene verificar

1. Orden. Cada miembro ve sus propios mensajes en orden, pero no hay orden global en el grupo. El mensaje 41 puede procesarse antes que el 40 en otro worker. Si el orden importa por dispositivo, usa una estrategia por hash de topic donde exista o particiona los topics. La prueba de orden de mensajes muestra cómo medirlo.
2. Mensajes retenidos. Publica un mensaje retenido y después arranca un suscriptor compartido: no debe recibir nada. Los workers que necesiten un estado inicial deben obtenerlo de otra forma.
3. QoS mixto. El QoS entregado es el mínimo entre el QoS de publicación y el QoS concedido a cada miembro. Un miembro suscrito en QoS 0 pierde mensajes al caer, aunque los demás usen QoS 1.
4. Opciones de suscripción. No Local en una suscripción compartida es un error de protocolo. Las demás opciones se explican en las opciones de suscripción MQTT.
5. Escalar el grupo. Añade un cuarto worker durante la ejecución: debería empezar a recibir su parte de inmediato. Quita uno: los demás absorben la carga.
6. Miembros persistentes. Con Session Expiry mayor que 0, un miembro desconectado conserva su sitio en el grupo, y algunos brokers siguen enviándole mensajes que se acumulan mientras está offline. Decide si los workers deben ser persistentes.

Probar la ganancia de throughput

Las suscripciones compartidas existen para escalar consumidores. Mide que el escalado es real: ejecuta el script con 1, 2 y 4 workers que esperen un tiempo fijo de procesamiento por mensaje y compara el tiempo total para vaciar una ráfaga. El throughput debería crecer casi linealmente hasta que el broker o la red pongan el límite. Una prueba de fan-in es el complemento natural: muchos publicadores alimentando un grupo.

Ver el reparto en directo

Abre el cliente MQTT online en dos pestañas del navegador, suscribe ambas al mismo filtro $share/... y publica desde una tercera pestaña. Verás el reparto en tiempo real, una buena comprobación antes de ejecutar el script.

Publicar trabajos de prueba desde el navegador

Una prueba de suscripciones compartidas lleva pocos minutos y responde a las preguntas que importan en producción: se procesa cada trabajo, se reparte de verdad la carga y qué pasa con el trabajo en curso cuando un worker muere. Mantén la ejecución de failover en tu suite de CI y repítela cada vez que cambies de broker o de configuración de sesión.

Preguntas frecuentes