Publicar y suscribirse desde el dispositivo

01 · Punto de partida

Del ejemplo al producto

El ejemplo del tema 25 cabe en veinte líneas: configurar el broker, conectar, suscribirse y publicar en el bucle. Funciona en la mesa de trabajo, con un broker local y un solo dispositivo. En una flota hay que responder preguntas que el ejemplo no se hace: ¿qué pasa si el broker rechaza la conexión? ¿Si la red se corta a mitad de una publicación? ¿Si el reinicio del broker deja a 20.000 equipos reconectándose al mismo tiempo? ¿Si llega una orden mientras el firmware está ocupado?

Es la misma disciplina que ya aplicaste con Wi-Fi (tema 14), HTTP (tema 19) y las fallas de red (tema 18), ahora con las particularidades de MQTT: una sesión con estado, una suscripción que hay que renovar y un callback que se ejecuta cuando el firmware menos lo espera.

Los fragmentos de C++ usan la biblioteca PubSubClient en un ESP32; los nombres corresponden a esa biblioteca y pueden variar entre versiones. Los números del laboratorio son resultados de una simulación con un generador pseudoaleatorio de semilla fija.

02 · Estados

El ciclo de vida del cliente

Un cliente MQTT atraviesa estados bien definidos, y conviene programarlo como una máquina de estados (la herramienta que ya conocés de Sistemas Embebidos) y no como una secuencia lineal de llamadas:

Dos propiedades de diseño importan por encima de todo. La primera: el cliente nunca bloquea. En la máquina de estados, cada paso hace un trabajo corto y devuelve el control; las esperas se miden con el reloj (millis()), no con delay. La segunda: ningún estado se supone permanente. «Operativo» puede terminar en cualquier momento, por un corte de red o una expulsión del broker, y el firmware debe notarlo y volver a empezar sin intervención.

La biblioteca más usada en Arduino tiene una trampa conocida que ilustra esto: su llamada de conexión es bloqueante y, si el broker no responde, puede quedarse esperando durante un tiempo largo (del orden de la decena de segundos por defecto). En un firmware que además atiende sensores o un botón, esa espera es un congelamiento. Se mitiga acortando los tiempos máximos y haciendo los intentos poco frecuentes.

03 · La conexión

Conectar bien

El identificador del cliente

Cada conexión lleva un identificador de cliente que debe ser único en el broker. Si dos clientes se conectan con el mismo, el broker desconecta al primero; este se reconecta, desconecta al segundo, y ambos entran en un bucle que consume recursos y nunca se estabiliza. Un identificador fijo copiado en todo el código ("sensor") es el origen de este error clásico. La solución es derivarlo de algo propio de cada equipo, como la dirección MAC o el número de serie del chip: camara07-a1b2c3.

Credenciales y cifrado

Usuario y clave (o certificado) propios de cada dispositivo, guardados en memoria no volátil y no en el código; conexión por TLS (puerto 8883) con verificación del certificado del broker, como en los temas 19 y 26. El nombre de usuario suele coincidir con el identificador del dispositivo, lo que permite a las listas de acceso del broker aplicar un patrón por usuario.

Interpretar la respuesta

El broker contesta el CONNECT con un CONNACK que lleva un código. Cada uno pide una reacción distinta:

Esta tabla es el equivalente, para MQTT, de la de códigos HTTP del tema 19: los errores 1, 2, 4 y 5 son permanentes y no se reintentan igual; el 3 y los fallos de red son transitorios. La biblioteca suele exponer el estado de la conexión como un número (por ejemplo, state()) que mezcla errores de red con esos códigos; hay que leerlo y decidir.

04 · Enviar datos

Publicar

Publicar parece lo más simple: una llamada con el tópico y la carga. Pero hay cuatro cuidados que separan un cliente confiable de uno que pierde datos sin avisar.

• El tamaño del búfer. Muchas bibliotecas reservan un búfer de tamaño fijo para armar cada paquete, y por defecto es chico (la más difundida usa 256 bytes). Si el paquete (encabezado, tópico y carga) lo excede, la publicación falla, a veces en silencio. Hay que dimensionarlo para el mensaje más grande que enviarás o recibirás, y comprobar el valor que devuelve la llamada.
• El resultado. La llamada devuelve si el mensaje pudo enviarse. Con calidad de servicio 0 eso no significa que haya llegado (tema 28), pero sí que no se perdió por un error local. Un firmware que ignora el resultado no se entera de que dejó de publicar.
• El patrón de envío. Hay tres formas, y se pueden combinar: periódica (una lectura cada cierto tiempo), por cambio (solo cuando el valor varía más que cierta histéresis, que ahorra tráfico), y por evento (alarmas y cambios de estado, que salen de inmediato).
• El contenido. Una carga compacta y autoexplicativa (tema 12): valor en entero escalado, momento de la medición y, si hace falta, identificador de secuencia. El identificador del dispositivo ya va en el tópico y no se repite.

Y una regla de convivencia con el resto del firmware: publicar no debe esperar a nada. Si la conexión no está operativa, la lectura se guarda en el buffer local (tema 18) y se publica cuando vuelva, con su marca de tiempo original.

05 · Mensajes entrantes

Recibir y obedecer

Los mensajes de los tópicos suscritos llegan a una función que el firmware registra, el callback. Lo que se haga adentro es crítico, porque se ejecuta en medio de la atención de la red:

• Debe ser corto. No esperar, no hacer cálculos largos, no dormir. Mientras el callback corre, el cliente no atiende el keepalive ni otros paquetes.
• No debe publicar con calidad alta ni hacer operaciones de red pesadas dentro de sí mismo en muchas bibliotecas. El patrón seguro es copiar el mensaje a una variable o una cola, marcar que hay algo pendiente y procesarlo en el bucle principal.
• El mensaje es un dato externo. Se valida como se vio en el tema 21: formato, tipos, rangos y vigencia. La carga llega como bytes sin terminador: hay que respetar su largo y no tratarla como una cadena de C sin cuidado (un desborde del búfer es una vulnerabilidad).
• Las órdenes son idempotentes y vencen (tema 11): un reenvío no duplica la acción, una orden vieja se descarta.
• Se confirma publicando el resultado en un tópico de respuestas, con el identificador de la orden.

La otra mitad del trabajo está en el bucle principal: hay que llamar a la función de atención de la biblioteca (loop) seguido, bastante más rápido que el keepalive. Si el firmware se queda 80 segundos en una tarea con un keepalive de 60, el broker lo da por caído a los 90 y expulsa la conexión. Cualquier espera larga del programa es una amenaza para la sesión.

Y una trampa de la reconexión: las suscripciones se piden dentro del código de conexión, no una sola vez al arrancar. Con una sesión limpia (la opción predeterminada), el broker olvida las suscripciones al caerse la conexión; si el firmware reconecta pero no se vuelve a suscribir, queda «conectado y sordo». El síntoma es desconcertante: el broker lo ve vivo, el dispositivo publica sin problemas, y no recibe ninguna orden.

06 · La tormenta

Reconectar sin tumbar al broker

En el tema 5 viste el reintento con espera creciente y retardo aleatorio para un dispositivo. En una flota MQTT el problema escala: cuando el broker se reinicia, todos los clientes pierden la conexión a la vez y tratan de reconectar, y conectar cuesta caro al broker (el apretón de manos de TLS consume procesador). A esto se le llama tormenta de reconexión o thundering herd, y puede mantener caído a un broker que ya había vuelto, porque cada oleada de intentos lo satura antes de que atienda a nadie.

Hay tres ingredientes que se combinan para evitarlo:

• Espera creciente entre intentos fallidos, con un tope (por ejemplo, 60 segundos), para que los equipos que no logran entrar se calmen.
• Retardo aleatorio (jitter) dentro de cada espera, para que los equipos no reintenten todos en el mismo segundo. Sin aleatoriedad, la espera creciente solo espacia las oleadas, no las rompe.
• Arranque escalonado: el primer intento de reconexión, y el de después de un reinicio, no es inmediato sino que espera un tiempo aleatorio (por ejemplo, de 0 a 30 segundos). Es lo que reduce el pico inicial.

El laboratorio de este tema simula el caso con cifras: miles de dispositivos que intentan conectar a la vez a un broker que acepta una cantidad limitada de conexiones nuevas por segundo. Mirá cómo cambia el tiempo de recuperación y el pico de intentos según la estrategia.

07 · Todo junto

Un esqueleto robusto en C++

#include <WiFi.h>
#include <PubSubClient.h>
#include "secretos.h"                             // USUARIO, CLAVE, SERVIDOR_MQTT

WiFiClient red;                                   // en producción: WiFiClientSecure, puerto 8883, con CA
PubSubClient mqtt(red);

const char *TOPICO_COMANDOS = "casa/camara07/comandos";
const char *TOPICO_DATOS    = "casa/camara07/temperatura";

char idCliente[32];
unsigned long proximoIntento = 0;                 // cuándo se permite volver a intentar
int fallos = 0;
char pendiente[200];                              // un mensaje recibido a la espera de procesarse
volatile bool hayPendiente = false;

void alMensaje(char *topico, byte *carga, unsigned int largo) {
  if (hayPendiente || largo >= sizeof(pendiente)) return;   // sin lugar: se descarta (o se cuenta)
  memcpy(pendiente, carga, largo);                // copiar y volver enseguida
  pendiente[largo] = 0;
  hayPendiente = true;
}

void gestionarConexion() {
  if (mqtt.connected()) { fallos = 0; return; }
  if (millis() < proximoIntento) return;          // todavía esperando: no bloquear
  if (mqtt.connect(idCliente, USUARIO, CLAVE)) {
    mqtt.subscribe(TOPICO_COMANDOS);              // se renueva en CADA conexión
    fallos = 0;
  } else {
    int tope = min(60, 1 << min(fallos, 6));      // 1, 2, 4, ... 60 segundos
    proximoIntento = millis() + (1 + esp_random() % tope) * 1000UL;   // espera creciente con retardo aleatorio
    fallos++;
    Serial.printf("MQTT falló (estado %d); próximo intento en hasta %d s\n", mqtt.state(), tope);
  }
}

void setup() {
  Serial.begin(115200);
  // ... conectar el Wi-Fi ...
  snprintf(idCliente, sizeof(idCliente), "camara07-%04x", (uint16_t)(ESP.getEfuseMac() >> 32));   // único por equipo
  mqtt.setServer(SERVIDOR_MQTT, 1883);
  mqtt.setCallback(alMensaje);
  mqtt.setKeepAlive(60);
  mqtt.setSocketTimeout(5);                       // no esperar 15 s al broker
  mqtt.setBufferSize(512);                        // el mensaje más grande que se espera, con margen
  proximoIntento = millis() + esp_random() % 30000;   // arranque escalonado
}

void loop() {
  gestionarConexion();
  mqtt.loop();                                    // atiende paquetes y keepalive: llamar seguido
  if (hayPendiente) {                             // procesar fuera del callback
    procesarOrden(pendiente);                     // validar, ejecutar, confirmar
    hayPendiente = false;
  }
  static unsigned long ultimo = 0;
  if (mqtt.connected() && millis() - ultimo > 60000) {
    ultimo = millis();
    if (!mqtt.publish(TOPICO_DATOS, "43")) Serial.println("Publicación fallida");
  }
}

Cada pieza del tema aparece acá: el identificador único derivado del chip, el callback que copia y vuelve, el procesamiento en el bucle, la suscripción dentro de la conexión, la espera creciente con retardo aleatorio, el arranque escalonado, el búfer ampliado, el tiempo máximo de conexión acortado y el resultado de publish revisado. La publicación solo ocurre con la conexión operativa; con un buffer local para lo que no pudo enviarse (tema 18) el firmware estaría completo.

08 · Analogía con programación

Simular una reconexión masiva en Python

El programa simula el momento en que un broker vuelve a estar disponible y todos los dispositivos intentan conectarse a la vez. El broker acepta como máximo una cantidad de conexiones nuevas por segundo; los intentos que exceden esa capacidad fallan. Cada dispositivo que falla reintenta según una de cuatro estrategias: una espera fija de 5 segundos; una espera creciente (1, 2, 4… 60 segundos) igual para todos; la creciente con un retardo aleatorio; y la creciente con retardo aleatorio más un arranque escalonado en el que el primer intento se reparte en 30 segundos. Un generador pseudoaleatorio de semilla fija hace los experimentos repetibles. Mide cuánto tarda en conectarse el 95 % y el 100 % de la flota, el mayor número de intentos en un segundo y los intentos por equipo.

HORIZONTE_S = 3600          # se deja de simular a la hora


class Aleatorio:
    """Generador congruencial: da siempre la misma secuencia, para repetir el experimento."""
    def __init__(self, semilla=777):
        self.estado = semilla

    def uniforme(self):
        self.estado = (self.estado * 1664525 + 1013904223) & 0xFFFFFFFF
        return self.estado / 4294967296


def espera(estrategia, fallos, azar):
    if estrategia == "fija":
        return 5
    tope = min(60, 2 ** fallos)                       # 1, 2, 4, 8, 16, 32, 60, 60...
    if estrategia == "creciente":
        return tope
    return 1 + int(azar.uniforme() * tope)            # creciente con retardo aleatorio ("jitter"), con o sin arranque escalonado


def simular(dispositivos, capacidad, estrategia):
    """Todos los dispositivos intentan conectarse a la vez (el broker acaba de volver).
    El broker acepta como máximo 'capacidad' conexiones nuevas por segundo; las demás fallan y reintentan."""
    azar = Aleatorio()
    agenda = {}
    for d in range(dispositivos):
        inicio = int(azar.uniforme() * 30) if estrategia == "jitter + arranque escalonado" else 0
        agenda.setdefault(inicio, []).append(d)           # el arranque escalonado reparte el primer intento en 30 s
    fallos = [0] * dispositivos
    intentos_totales, pico, conectados, t95, t100 = 0, 0, 0, None, None
    for t in range(HORIZONTE_S):
        lista = agenda.pop(t, [])
        if not lista:
            continue
        intentos_totales += len(lista)
        pico = max(pico, len(lista))
        aceptados, rechazados = lista[:capacidad], lista[capacidad:]
        conectados += len(aceptados)
        for d in rechazados:
            fallos[d] += 1
            agenda.setdefault(t + espera(estrategia, fallos[d] - 1, azar), []).append(d)
        if t95 is None and conectados >= 0.95 * dispositivos:
            t95 = t
        if conectados == dispositivos:
            t100 = t
            break
    return t95, t100, pico, intentos_totales / dispositivos


print("dispositivos  capacidad  estrategia                      95 % conectado   todos conectados   pico de intentos/s   intentos por equipo")
for dispositivos, capacidad in ((5_000, 100), (20_000, 100), (20_000, 500)):
    for estrategia in ("fija", "creciente", "creciente + jitter", "jitter + arranque escalonado"):
        t95, t100, pico, medio = simular(dispositivos, capacidad, estrategia)
        a = "—" if t95 is None else f"{t95} s"
        b = "no en 1 h" if t100 is None else f"{t100} s"
        print(f"{dispositivos:>11,} {capacidad:>8}/s  {estrategia:<30} {a:>14} {b:>18} {pico:>20,} {medio:>19.1f}")

Resultado con los valores iniciales. Con 5000 dispositivos y un broker que acepta 100 conexiones por segundo (el mínimo teórico son 50 segundos): la espera fija conecta a todos en 245 s con 25,5 intentos por equipo; la creciente sin aleatoriedad, en 2643 s (44 minutos); la creciente con retardo aleatorio, en 88 s con 6,0 intentos; y con arranque escalonado, en 96 s, con solo 3,2 intentos por equipo y un pico de 559 intentos por segundo en lugar de 5000. Con 20.000 dispositivos y el mismo broker: la fija tarda 995 s y 100,5 intentos por equipo; la creciente sin aleatoriedad no se recupera en una hora; la aleatoria, 241 s; y la escalonada, 235 s con un pico de 3771. Si el broker acepta 500 por segundo, con 20.000 dispositivos: fija, 195 s; creciente, 2043 s; con retardo aleatorio, 86 s; escalonada, 89 s con 1583 de pico y 2,3 intentos por equipo.

Tres conclusiones. Una: la espera creciente sin aleatoriedad es la peor estrategia: espacia las oleadas pero cada una conecta solo 100 equipos, de modo que con 20.000 no termina en una hora. Otra: el retardo aleatorio acerca el tiempo de recuperación al mínimo teórico (86 s contra 40 s para 20.000 equipos a 500 por segundo) y reduce los intentos desperdiciados hasta 11 veces (100,5 contra 9,0 por equipo con 20.000 dispositivos a 100 por segundo), que en un equipo con batería son energía. Y la tercera: el arranque escalonado es lo único que baja el pico inicial, que sigue siendo la flota entera si todos intentan en el mismo segundo.

09 · Exploración interactiva

El broker vuelve y todos entran, de extremo a extremo

Este laboratorio ejecuta la misma simulación del código anterior, con el mismo generador de números pseudoaleatorios. Elegí cuántos dispositivos hay, cuántas conexiones nuevas por segundo acepta el broker y qué estrategia de reintento usa el firmware. Las barras comparan las cuatro estrategias en el tiempo hasta que todos están conectados (escala logarítmica); en rojo, las que no se recuperan en una hora. Al lado de cada barra figura el pico de intentos por segundo.

Recorrido guiado

1. Con los valores iniciales (20.000 dispositivos, 500 por segundo, creciente con retardo aleatorio), simulá. El 95 % entra en 52 s y toda la flota en 86 s, con 5,7 intentos por equipo y un pico de 20.000 en el primer segundo: «Se recupera bien».
2. Cambiá a Espera fija de 5 s: 195 s (3,3 minutos) y 20,5 intentos por equipo, 3,6 veces más que con el retardo aleatorio. Funciona, pero cada oleada sincronizada es una avalancha y gasta energía.
3. Probá Espera creciente sin aleatoriedad: 2043 s, 34 minutos, con la misma cantidad de intentos que la fija. Las oleadas se espacian pero no se dispersan, y cada una conecta solo 500 equipos.
4. Bajá la capacidad del broker a 100 por segundo con la creciente: no se recupera en una hora. Con la fija son 995 s; con el retardo aleatorio, 241 s.
5. Elegí arranque escalonado con 20.000 y 500 por segundo: el pico de intentos baja de 20.000 a 1583 y los intentos por equipo a 2,3, con la recuperación casi igual (89 s). Es la estrategia que más cuida al broker y a la batería.
6. Pasá a 5000 dispositivos y 100 por segundo con arranque escalonado: 96 s, pico de 559 y 3,2 intentos por equipo. El mínimo teórico es de 50 s; con aleatoriedad no se llega al mínimo, pero se queda en el doble.

10 · Método de trabajo

Lista de verificación del cliente MQTT

1. Identificador único por equipo, derivado de la MAC o del número de serie.
2. Credencial propia, TLS con verificación del certificado y datos secretos fuera del código.
3. Máquina de estados no bloqueante: ninguna espera con delay; los tiempos se miden con el reloj.
4. Interpretar el código de la conexión y distinguir lo transitorio de lo permanente.
5. Suscribirse dentro de cada conexión, y revisar el resultado.
6. Búfer del tamaño del mensaje más grande, y verificar el resultado de cada publicación.
7. Callback breve: copiar el mensaje y procesarlo en el bucle principal, validado.
8. Atender la red seguido: la función de la biblioteca se llama mucho más rápido que el keepalive.
9. Reconexión con espera creciente, retardo aleatorio y arranque escalonado.
10. Buffer local para lo que no se pudo publicar, con marcas de tiempo originales (tema 18).
11. Pruebas de caos: reiniciar el broker con la flota conectada, cortar el Wi-Fi, bloquear el puerto, enviar mensajes enormes, dos dispositivos con el mismo identificador.

La prueba que más revela: reiniciar el broker de prueba con varios dispositivos conectados y medir cuánto tardan en volver, cuántos intentos hacen y si recuperan las suscripciones. Un cliente que pasa esa prueba está listo para una flota.

11 · Comprensión

Confusiones frecuentes

12 · Práctica guiada

Predecí antes de comprobar

1. Qué hacer con el CONNACK

Indicá si reintentarías y cómo para cada código: (a) 3; (b) 4; (c) 5; (d) 2.

2. El búfer

Un tópico de 25 caracteres y una carga de 300 bytes. Con un búfer de 256, ¿entra el paquete? ¿Cuánto mide?

3. Keepalive

Keepalive de 60 s. El firmware hace una tarea de 100 segundos sin llamar a la función de atención de la red. ¿Qué pasa?

4. Espera creciente

Con tope = min(60, 2^fallos) segundos, ¿cuál es la espera máxima tras cada uno de los primeros 8 fallos (fallos = 0, 1, …, 7)? ¿Y en promedio, con retardo aleatorio uniforme?

5. Tiempo mínimo de recuperación

20.000 dispositivos y un broker que acepta 100 conexiones por segundo. ¿Cuál es el mínimo teórico para que todos estén conectados? ¿Cuánto tardó la espera fija y el retardo aleatorio en la simulación?

6. Conectado y sordo

Un dispositivo publica con normalidad pero hace semanas que no recibe órdenes. El broker lo muestra conectado. ¿Qué revisarías primero?

13 · Síntesis

Ideas para recordar

¿Listo para el tema 28?

Antes de avanzar, reiniciá un broker de prueba con tres dispositivos conectados y registrá, para cada uno: cuánto tardó en volver, cuántos intentos hizo y si recuperó las suscripciones. Revisá que el identificador sea único, que el búfer alcance para tu mensaje más grande y que ningún delay del bucle supere la mitad del keepalive.

En el próximo tema estudiamos las garantías de entrega: QoS, mensajes retenidos y última voluntad, con las que MQTT resuelve las órdenes perdidas durante una desconexión y la detección de dispositivos que callan.