MQTT y Redis: cachea el último estado de cada dispositivo

Para cachear en Redis el último estado de tus dispositivos MQTT, ejecuta un pequeño suscriptor que escuche tus topics de telemetría y escriba cada mensaje en un hash de Redis con el ID del dispositivo como clave. Tus APIs, dashboards y tareas en segundo plano leen entonces el estado actual con un solo HGETALL, sin abrir una conexión MQTT ni esperar al siguiente mensaje.

Este tutorial construye ese suscriptor en Python, lo protege frente a actualizaciones desordenadas, añade un Redis Stream para el historial reciente y compara el enfoque con los mensajes retenidos y con Redis Pub/Sub.

Cuándo tiene sentido una caché en Redis

MQTT ya tiene una función de "último valor": los mensajes retenidos. Un nuevo suscriptor recibe de inmediato el último mensaje retenido de cada topic. Entonces, ¿para qué añadir Redis?

Si solo los clientes MQTT necesitan el último valor, los mensajes retenidos bastan. En cuanto código que no es MQTT necesita el estado, una caché en Redis es más simple y rápida.

El modelo de datos

Mantenlo aburrido y predecible:

• device:<id>: un hash con los últimos campos: temp, humidity, battery, ts (timestamp del dispositivo), topic, seen (cuándo lo escribió el puente).
• devices:online: un set opcional de IDs de dispositivo, mantenido a partir de topics de estado.
• telemetry: un stream opcional con mensajes en bruto recientes para replay y depuración.

Los dispositivos publican JSON en testmqtt/redis/<device>/state, por ejemplo {"ts": 1760700000123, "temp": 21.4, "battery": 87}. Un timestamp del dispositivo (o un número de secuencia) es lo que te permitirá rechazar actualizaciones obsoletas. Consulta payloads JSON en Python para consejos de parseo.

El suscriptor

Este puente usa paho-mqtt 2.x y redis-py. El script Lua escribe el hash solo si el timestamp entrante es más nuevo que el guardado, y renueva un TTL para que los dispositivos silenciosos desaparezcan solos:

import json, os, time
import paho.mqtt.client as mqtt
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# KEYS[1] = clave del dispositivo; ARGV = ts, ttl y luego pares campo/valor
UPSERT_IF_NEWER = r.register_script("""
local cur = redis.call('HGET', KEYS[1], 'ts')
if cur and tonumber(cur) >= tonumber(ARGV[1]) then
  return 0
end
redis.call('HSET', KEYS[1], 'ts', ARGV[1], unpack(ARGV, 3))
redis.call('EXPIRE', KEYS[1], ARGV[2])
return 1
""")

TTL_SECONDS = 900   # olvidar dispositivos que llevan 15 minutos en silencio

def on_connect(client, userdata, flags, reason_code, properties):
    print("conectado:", reason_code)
    client.subscribe("testmqtt/redis/+/state", qos=1)

def on_message(client, userdata, msg):
    device = msg.topic.split("/")[2]
    try:
        data = json.loads(msg.payload)
    except ValueError:
        print("se ignora payload no JSON en", msg.topic)
        return

    ts = int(data.pop("ts", time.time() * 1000))
    fields = []
    for k, v in data.items():
        plain = isinstance(v, (str, int, float)) and not isinstance(v, bool)
        fields += [k, v if plain else json.dumps(v)]   # bool/None/objetos como JSON
    fields += ["topic", msg.topic, "seen", int(time.time() * 1000)]

    written = UPSERT_IF_NEWER(keys=[f"device:{device}"], args=[ts, TTL_SECONDS, *fields])
    if not written:
        print(f"actualización obsoleta de {device} ignorada (ts={ts})")

    # historial corto opcional, limitado a ~100k entradas
    r.xadd("telemetry", {"device": device, "payload": msg.payload}, maxlen=100_000, approximate=True)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="redis-state-bridge")
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883)
client.loop_forever()

Notas sobre el diseño:

• Compare-and-set atómico. Los scripts Lua se ejecutan de forma atómica en Redis, así que dos instancias del puente (o una reentrega QoS 1) no pueden competir y escribir un estado más antiguo.
• Campos planos. Strings y números se guardan tal cual; booleanos, nulos y objetos anidados se guardan como strings JSON. Los hashes de Redis tienen un solo nivel: no luches contra eso.
• El TTL como señal de desconexión. Si una clave expira, el dispositivo no ha reportado en TTL_SECONDS. Para una señal explícita, combínalo con un mensaje Last Will en un topic de estado.
• Los duplicados no hacen daño. Escribir dos veces el mismo estado no cambia nada, lo que hace que este destino sea idempotente por naturaleza.

Leer el estado

Cualquier servicio puede responder ahora "¿qué está haciendo el dispositivo X en este momento?" sin MQTT:

redis-cli HGETALL device:sensor-1
redis-cli HGET device:sensor-1 temp
redis-cli TTL device:sensor-1
redis-cli --scan --pattern 'device:*'

Para un dashboard con actualizaciones en vivo, sirve la carga inicial de la página desde un endpoint REST respaldado por Redis y deja que el navegador se suscriba a MQTT por WebSockets seguros (puerto 8084, path /mqtt) para los cambios posteriores. La caché responde al "ahora"; MQTT entrega el "siguiente".

Redis Streams para el historial reciente

La línea XADD mantiene un log acotado de mensajes en bruto. Los streams admiten consumer groups, así que varios workers pueden procesar actualizaciones con confirmaciones, y puedes reproducir un rango de tiempo al depurar:

redis-cli XLEN telemetry
redis-cli XREVRANGE telemetry + - COUNT 5

redis-cli XGROUP CREATE telemetry alerts $ MKSTREAM
redis-cli XREADGROUP GROUP alerts worker-1 COUNT 10 BLOCK 5000 STREAMS telemetry '>'

Mantén el stream corto (MAXLEN): vive en memoria. Para historial a largo plazo, escribe en una base de datos de series temporales o en Postgres; consulta MQTT a PostgreSQL y retención y almacenamiento de datos MQTT.

MQTT vs Redis Pub/Sub

Redis también tiene Pub/Sub, y es tentador saltarse MQTT por completo. No son intercambiables:

• Redis Pub/Sub es fire-and-forget entre procesos conectados a Redis en ese momento. No hay QoS, ni valor retenido, ni Last Will, ni un modelo de ACL por topic pensado para dispositivos, y los clientes deben mantener una conexión con Redis.
• MQTT está diseñado para dispositivos en redes poco fiables: QoS 1 y 2, sesiones persistentes, mensajes retenidos, keep alive, Last Will y clientes ligeros para microcontroladores.

Un reparto habitual es MQTT en el edge y Redis dentro del backend. La comparación completa está en MQTT vs Redis Pub/Sub.

Pruébalo

Arranca Redis en local y ejecuta el puente:

docker run -d --name redis -p 6379:6379 redis:7
python mqtt_redis_state.py

Publica un estado nuevo y después uno más antiguo con un timestamp menor:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 2000, "temp": 22.5, "battery": 90}'

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/redis/sensor-1/state -m '{"ts": 1000, "temp": 19.0, "battery": 95}'

redis-cli HGETALL device:sensor-1

temp debe seguir siendo 22.5 y el puente debe registrar actualización obsoleta de sensor-1 ignorada. También puedes enviar los payloads a mano desde el cliente MQTT online y observar redis-cli MONITOR para ver cada comando que emite el puente.

Publica estados de prueba desde el navegador

Una caché de estado en Redis es una de las mejoras más baratas para un sistema MQTT: un suscriptor, un hash por dispositivo y todos los servicios con acceso instantáneo al "ahora". Añade la comprobación de timestamp desde el primer día, deja el historial fuera del hash y permite que MQTT siga haciendo lo que mejor sabe: mover las actualizaciones.

Preguntas frecuentes