MQTT et Redis : mettre en cache le dernier état des appareils

Pour mettre en cache dans Redis le dernier état de vos appareils MQTT, exécutez un petit abonné qui écoute vos topics de télémétrie et écrit chaque message dans un hash Redis dont la clé est l'ID de l'appareil. Vos API, dashboards et tâches de fond lisent alors l'état courant avec un seul HGETALL, sans ouvrir de connexion MQTT ni attendre le prochain message.

Ce tutoriel construit cet abonné en Python, le protège contre les mises à jour dans le désordre, ajoute un Redis Stream pour l'historique récent et compare l'approche aux messages retenus et à Redis Pub/Sub.

Quand un cache Redis a du sens

MQTT possède déjà une fonction « dernière valeur » : les messages retenus. Un nouvel abonné reçoit immédiatement le dernier message retenu de chaque topic. Alors pourquoi ajouter Redis ?

Si seuls des clients MQTT ont besoin de la dernière valeur, les messages retenus suffisent. Dès que du code non MQTT a besoin de l'état, un cache Redis est plus simple et plus rapide.

Le modèle de données

Restez simple et prévisible :

• device:<id> : un hash avec les derniers champs : temp, humidity, battery, ts (timestamp de l'appareil), topic, seen (moment où le bridge l'a écrit).
• devices:online : un set optionnel d'ID d'appareils, alimenté par des topics de statut.
• telemetry : un stream optionnel des messages bruts récents, pour le replay et le débogage.

Les appareils publient du JSON sur testmqtt/redis/<device>/state, par exemple {"ts": 1760700000123, "temp": 21.4, "battery": 87}. Un timestamp fourni par l'appareil (ou un numéro de séquence) est ce qui vous permettra ensuite de rejeter les mises à jour périmées. Consultez les payloads JSON en Python pour des conseils de parsing.

L'abonné

Ce bridge utilise paho-mqtt 2.x et redis-py. Le script Lua n'écrit le hash que si le timestamp entrant est plus récent que celui stocké, et renouvelle un TTL pour que les appareils muets disparaissent d'eux-mêmes :

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

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

# KEYS[1] = clé de l'appareil ; ARGV = ts, ttl, puis paires champ/valeur
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   # oublier les appareils muets depuis 15 minutes

def on_connect(client, userdata, flags, reason_code, properties):
    print("connecté :", 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("payload non JSON ignoré sur", 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/objets en 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"mise à jour périmée pour {device} ignorée (ts={ts})")

    # historique court optionnel, plafonné à ~100k entrées
    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()

Remarques sur la conception :

• Compare-and-set atomique. Les scripts Lua s'exécutent de façon atomique dans Redis : deux instances du bridge (ou une relivraison QoS 1) ne peuvent pas se concurrencer et écrire un état plus ancien.
• Champs plats. Les chaînes et les nombres sont stockés tels quels ; booléens, valeurs nulles et objets imbriqués sont stockés en chaînes JSON. Les hashes Redis n'ont qu'un niveau, ne luttez pas contre ça.
• Le TTL comme signal hors ligne. Si une clé expire, l'appareil n'a rien envoyé depuis TTL_SECONDS. Pour un signal explicite, combinez-le avec un message Last Will sur un topic de statut.
• Les doublons sont sans effet. Écrire deux fois le même état ne change rien, ce qui rend cette destination naturellement idempotente.

Lire l'état

N'importe quel service peut désormais répondre à « que fait l'appareil X en ce moment ? » sans 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:*'

Pour un dashboard en temps réel, servez le chargement initial via un endpoint REST adossé à Redis, puis laissez le navigateur s'abonner à MQTT via WebSockets sécurisés (port 8084, chemin /mqtt) pour les changements suivants. Le cache répond au « maintenant », MQTT livre la « suite ».

Redis Streams pour l'historique récent

La ligne XADD conserve un log plafonné des messages bruts. Les streams prennent en charge les consumer groups : plusieurs workers peuvent traiter les mises à jour avec acquittement, et vous pouvez rejouer une plage de temps pendant le débogage :

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 '>'

Gardez le stream court (MAXLEN) : il vit en mémoire. Pour un historique long terme, écrivez dans une base de séries temporelles ou dans Postgres ; voir MQTT vers PostgreSQL et rétention et stockage des données MQTT.

MQTT vs Redis Pub/Sub

Redis dispose aussi d'un Pub/Sub, et il est tentant de se passer complètement de MQTT. Les deux ne sont pas interchangeables :

• Redis Pub/Sub est du fire-and-forget entre processus connectés à Redis à cet instant. Pas de QoS, pas de valeur retenue, pas de Last Will, pas de modèle d'ACL par topic pensé pour des appareils, et les clients doivent garder une connexion Redis.
• MQTT est conçu pour des appareils sur des réseaux peu fiables : QoS 1 et 2, sessions persistantes, messages retenus, keep alive, Last Will et clients légers pour microcontrôleurs.

Une répartition courante : MQTT en périphérie, Redis dans le backend. La comparaison complète se trouve dans MQTT vs Redis Pub/Sub.

Tester

Démarrez Redis en local et lancez le bridge :

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

Publiez un état récent, puis un plus ancien avec un timestamp inférieur :

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 doit toujours valoir 22.5, et le bridge doit journaliser mise à jour périmée pour sensor-1 ignorée. Vous pouvez aussi envoyer les payloads à la main depuis le client MQTT en ligne et suivre redis-cli MONITOR pour voir chaque commande émise par le bridge.

Publier des états de test depuis le navigateur

Un cache d'état Redis est l'une des améliorations les moins coûteuses d'un système MQTT : un abonné, un hash par appareil, et chaque service accède instantanément au « maintenant ». Ajoutez la vérification de timestamp dès le premier jour, gardez l'historique hors du hash et laissez MQTT continuer à faire ce qu'il fait le mieux : transporter les mises à jour.

Questions fréquentes