Suscripciones compartidas MQTT: balanceo de carga entre consumidores

Las suscripciones compartidas MQTT permiten que varios clientes se suscriban como grupo, de modo que cada mensaje de un topic se entrega solo a un miembro de ese grupo en lugar de a todos. Al terminar este artículo entenderás la sintaxis $share/group/topic, cómo funciona la entrega cuando los workers entran y salen, cómo escalar con ella los consumidores del backend y cómo comprobar si tu broker la admite.

Por qué las suscripciones normales no escalan a los consumidores

En la publicación/suscripción estándar, cada suscriptor cuyo filtro coincide recibe su propia copia de cada mensaje. Es exactamente lo que quieres para los dashboards: cada pantalla muestra cada lectura. Y es exactamente lo que no quieres para el procesamiento en el backend. Si ejecutas tres instancias de un servicio que escribe telemetría en una base de datos, tres suscripciones normales significan que cada lectura se escribe tres veces.

Antes de MQTT 5, las alternativas eran incómodas: un único consumidor que se convertía en cuello de botella, particionar los topics a mano entre instancias o hacer un bridge de MQTT a un sistema de colas aparte. Las suscripciones compartidas resuelven el problema dentro del propio broker MQTT.

Cómo funcionan las suscripciones compartidas MQTT

Una suscripción compartida usa un formato de filtro especial, estandarizado en MQTT 5:

$share/<ShareName>/<TopicFilter>

• $share es el prefijo fijo que marca la suscripción como compartida.
• <ShareName> da nombre al grupo de consumidores. No debe contener /, + ni #.
• <TopicFilter> es un filtro normal, comodines incluidos.

Cada cliente que se suscribe al mismo $share/<ShareName>/<TopicFilter> se une al mismo grupo. Para cada mensaje coincidente, el broker elige un miembro del grupo y le entrega el mensaje. Los distintos nombres de grupo son grupos independientes, y los suscriptores normales del topic simple siguen recibiendo sus propias copias.

La especificación deja la elección del miembro en manos del broker. Las estrategias round-robin, aleatoria y "el menos ocupado" son todas habituales, así que no dependas de un orden concreto ni de que los mensajes de un dispositivo lleguen siempre al mismo worker.

Semántica de entrega que debes conocer

1. El orden es por miembro, no por topic. Cada worker ve sus propios mensajes en orden, pero dos lecturas consecutivas del mismo sensor pueden ir a workers distintos y procesarse en paralelo. Si el orden importa, incluye un número de secuencia o una marca de tiempo y resuélvelo en tu base de datos.
2. El QoS sigue aplicándose. Con QoS 1, si un worker se desconecta antes de confirmar un mensaje, los brokers suelen reenviarlo a la misma sesión cuando se reconecta, o encaminarlo a otro miembro; el comportamiento varía según el broker. Haz que los handlers sean idempotentes.
3. Los mensajes retenidos no se envían al suscribirse. La especificación establece que los mensajes retenidos no se entregan a una suscripción compartida cuando se crea, porque no estaría claro qué miembro debería recibirlos.
4. No Local está prohibido. Activar la opción de suscripción No Local en una suscripción compartida es un error de protocolo.
5. Las sesiones son por miembro. Cada worker tiene su propio client ID y su propia sesión. Una sesión persistente en un worker mantiene viva la suscripción de ese worker mientras está offline. Consulta las sesiones persistentes MQTT para ver cómo afecta eso al encolado.

Ejemplo: escalar workers en Python

El worker siguiente se une a un grupo llamado db-writers y procesa telemetría. Ejecútalo en dos o tres terminales con números de worker distintos para ver cómo se reparte la carga.

import sys
import json
import paho.mqtt.client as mqtt

WORKER = sys.argv[1] if len(sys.argv) > 1 else "1"
FILTER = "$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry"

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Worker {WORKER} connected: {reason_code}")
    client.subscribe(FILTER, qos=1)

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscribe rejected: {rc}")  # p. ej. Shared Subscriptions not supported
        else:
            print(f"Joined group, granted {rc}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    print(f"Worker {WORKER} got {msg.topic}: {data}")
    # escribe aquí en tu base de datos; mantenlo idempotente

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"db-writer-s7h3-{WORKER}",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Inicia dos workers:

python worker.py 1
python worker.py 2

Después publica una serie de lecturas de prueba:

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i sh-pub-s7h3 -q 1 \
    -t "connectmq/demo/s7h3/factory/line$((i % 3))/telemetry" \
    -m "{\"seq\": $i, \"temp\": 70}"
done

Si el broker admite suscripciones compartidas, los diez mensajes se reparten entre los workers en lugar de que cada worker imprima los diez. Sustituye s7h3 por tu propia cadena aleatoria; cualquiera puede leer el servidor MQTT público gratuito, así que publica solo datos ficticios.

La misma suscripción en MQTT.js tiene este aspecto:

const mqtt = require("mqtt");
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "db-writer-node-s7h3-1",
  protocolVersion: 5,
});
client.on("connect", () => {
  client.subscribe("$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry",
    { qos: 1 }, (err, granted) => console.log(err || granted));
});
client.on("message", (topic, payload) => console.log(topic, payload.toString()));

Cómo verificar el soporte de suscripciones compartidas

El soporte varía entre brokers y entre planes, así que compruébalo en lugar de darlo por hecho:

1. Lee el SUBACK. En MQTT 5, un broker que no admite suscripciones compartidas devuelve el reason code 0x9E (Shared Subscriptions not supported). El callback on_subscribe anterior lo imprime.
2. Revisa el CONNACK. Los brokers MQTT 5 pueden anunciar Shared Subscription Available = 0 en el CONNACK. Si la propiedad no está presente, se asume que hay soporte.
3. Haz la prueba de reparto. Dos miembros, diez mensajes. Si ambos miembros imprimen los diez, el broker ha tratado $share/... como un filtro de topic normal y no como un grupo.
4. Comprueba el plan. Los brokers MQTT gestionados suelen vincular la funcionalidad a un nivel concreto. En ConnectMQ, las suscripciones compartidas están incluidas en el plan Serverless (Paid) y superiores.
5. Prueba MQTT 3.1.1 por separado. Algunos brokers aceptan $share de clientes 3.1.1 como extensión; otros exigen MQTT 5. No dependas de ello salvo que tu broker lo documente.

Suscripciones compartidas frente a otros enfoques

Hay otras formas de repartir el trabajo, y conviene saber cuándo encaja cada una. El particionado manual de topics (el worker 1 se suscribe a line1 y el worker 2 a line2) ofrece un orden estricto por partición, pero exige reconfigurar cada vez que añades un worker o una línea. Un único consumidor que alimenta una cola interna mantiene el orden de forma sencilla, pero convierte ese proceso en un cuello de botella y en un punto único de fallo. Las suscripciones compartidas no necesitan coordinación entre workers y escalan simplemente arrancando más instancias, a costa del orden por topic entre workers. Para la mayoría de los casos de ingesta de telemetría y procesamiento de alertas, ese compromiso es el adecuado.

Patrones y errores comunes

• Un grupo por propósito. Usa un nombre de grupo por cada trabajo independiente (db-writers, alerting, exporter). Así, cada trabajo recibe cada mensaje exactamente una vez entre sus workers.
• Autoescalado. Añadir un worker consiste simplemente en conectar otro cliente con el mismo filtro; quitarlo es una desconexión. Da a cada instancia un client ID único, como se explica en las buenas prácticas del client ID de MQTT.
• Control de flujo. Combina las suscripciones compartidas con un Receive Maximum de MQTT 5 pequeño en cada worker para que a una instancia lenta no se le entregue una gran ráfaga de mensajes sin confirmar.
• Los comodines están bien. $share/group/factory/# reparte una rama completa. Las reglas de coincidencia son las habituales, explicadas en los comodines de MQTT.
• No lo uses para fan-out. Si todos los consumidores deben ver todos los mensajes, usa suscripciones normales.

¿Eres nuevo en MQTT en general? El tutorial de MQTT trata suscripciones, QoS y sesiones desde el principio.

Escala consumidores en ConnectMQ

ConnectMQ es un broker MQTT gestionado compatible con MQTT 3.1.1 y 5.0, con suscripciones compartidas disponibles a partir del plan Serverless (Paid), junto con QoS 2, mensajes de 1 MB y hasta 1.000 conexiones. Las opciones Dedicated y BYOC están disponibles cuando necesitas un clúster aislado. Regístrate y empieza con el plan Serverless gratuito sin tarjeta de crédito, y actualiza cuando tu backend necesite grupos de consumidores. Conecta tus workers con YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID.

Preguntas frecuentes

¿Cuál es la sintaxis de una suscripción compartida MQTT?

Usa $share/<ShareName>/<TopicFilter>, por ejemplo $share/workers/sensors/+/temperature. Todos los clientes que usan el mismo nombre de grupo y el mismo filtro forman un grupo.

¿Funcionan las suscripciones compartidas con MQTT 3.1.1?

Están definidas en MQTT 5. Algunos brokers aceptan la sintaxis $share de clientes 3.1.1 como extensión, pero debes confirmarlo en la documentación de tu broker antes de depender de ello.

¿Se entregan los mensajes retenidos a las suscripciones compartidas?

No. Cuando se crea una suscripción compartida, no se le envían los mensajes retenidos. Si un worker necesita el estado retenido, léelo con una suscripción normal aparte.

¿Cómo elige el broker qué suscriptor recibe un mensaje?

La especificación de MQTT lo deja en manos del broker. La selección round-robin y la aleatoria son habituales, así que nunca dependas de que un miembro concreto reciba un mensaje concreto.