MQTT e Redis: cache do último estado de cada dispositivo

Para fazer cache no Redis do último estado dos seus dispositivos MQTT, rode um pequeno assinante que escuta seus topics de telemetria e grava cada mensagem em um hash do Redis com o ID do dispositivo como chave. Suas APIs, dashboards e jobs em segundo plano leem então o estado atual com um único HGETALL, sem abrir uma conexão MQTT nem esperar a próxima mensagem.

Este tutorial constrói esse assinante em Python, protege-o contra atualizações fora de ordem, adiciona um Redis Stream para o histórico recente e compara a abordagem com mensagens retidas e com o Redis Pub/Sub.

Quando um cache no Redis faz sentido

O MQTT já tem um recurso de "último valor": as mensagens retidas. Um novo assinante recebe imediatamente a última mensagem retida de cada topic. Então por que adicionar o Redis?

Se só clientes MQTT precisam do último valor, mensagens retidas bastam. Assim que código que não fala MQTT precisa do estado, um cache no Redis é mais simples e mais rápido.

O modelo de dados

Mantenha tudo simples e previsível:

• device:<id>: um hash com os últimos campos: temp, humidity, battery, ts (timestamp do dispositivo), topic, seen (quando a ponte gravou).
• devices:online: um set opcional de IDs de dispositivo, mantido a partir de topics de status.
• telemetry: um stream opcional com mensagens brutas recentes para replay e depuração.

Os dispositivos publicam JSON em testmqtt/redis/<device>/state, por exemplo {"ts": 1760700000123, "temp": 21.4, "battery": 87}. Um timestamp vindo do dispositivo (ou um número de sequência) é o que permite rejeitar atualizações antigas depois. Veja payloads JSON em Python para dicas de parsing.

O assinante

Esta ponte usa paho-mqtt 2.x e redis-py. O script Lua grava o hash somente se o timestamp recebido for mais novo que o salvo, e renova um TTL para que dispositivos silenciosos sumam sozinhos:

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

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

# KEYS[1] = chave do dispositivo; ARGV = ts, ttl e depois 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   # esquecer dispositivos em silêncio há 15 minutos

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("ignorando payload não JSON em", 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"atualização antiga de {device} ignorada (ts={ts})")

    # histórico curto 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 o design:

• Compare-and-set atômico. Scripts Lua rodam de forma atômica no Redis, então duas instâncias da ponte (ou uma reentrega QoS 1) não conseguem disputar e gravar um estado mais antigo.
• Campos planos. Strings e números entram como estão; booleanos, nulos e objetos aninhados são gravados como strings JSON. Hashes do Redis têm um único nível; não brigue com isso.
• TTL como sinal de offline. Se uma chave expira, o dispositivo não reportou por TTL_SECONDS. Para um sinal explícito, combine com uma mensagem Last Will em um topic de status.
• Duplicatas são inofensivas. Gravar o mesmo estado duas vezes não muda nada, o que torna esse destino naturalmente idempotente.

Lendo o estado

Qualquer serviço agora responde "o que o dispositivo X está fazendo agora?" sem 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 um dashboard com atualizações ao vivo, sirva o carregamento inicial da página por um endpoint REST apoiado no Redis e deixe o navegador assinar o MQTT por WebSockets seguros (porta 8084, path /mqtt) para as mudanças seguintes. O cache responde o "agora"; o MQTT entrega o "próximo".

Redis Streams para o histórico recente

A linha XADD mantém um log limitado de mensagens brutas. Streams suportam consumer groups, então vários workers podem processar atualizações com confirmação, e você pode reprocessar um intervalo de tempo ao 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 '>'

Mantenha o stream curto (MAXLEN): ele fica em memória. Para histórico de longo prazo, grave em um banco de séries temporais ou no Postgres; veja MQTT para PostgreSQL e retenção e armazenamento de dados MQTT.

MQTT vs Redis Pub/Sub

O Redis também tem Pub/Sub, e é tentador pular o MQTT de vez. Os dois não são intercambiáveis:

• Redis Pub/Sub é fire-and-forget entre processos conectados ao Redis naquele momento. Não há QoS, valor retido, Last Will nem um modelo de ACL por topic pensado para dispositivos, e os clientes precisam manter uma conexão com o Redis.
• MQTT foi projetado para dispositivos em redes instáveis: QoS 1 e 2, sessões persistentes, mensagens retidas, keep alive, Last Will e clientes leves para microcontroladores.

Uma divisão comum é MQTT na borda e Redis dentro do backend. A comparação completa está em MQTT vs Redis Pub/Sub.

Teste

Suba o Redis localmente e rode a ponte:

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

Publique um estado novo e depois um mais antigo, com 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 precisa continuar 22.5, e a ponte deve registrar atualização antiga de sensor-1 ignorada. Você também pode enviar os payloads manualmente pelo cliente MQTT online e acompanhar redis-cli MONITOR para ver cada comando que a ponte executa.

Publique estados de teste pelo navegador

Um cache de estado no Redis é uma das melhorias mais baratas para um sistema MQTT: um assinante, um hash por dispositivo e todos os serviços com acesso instantâneo ao "agora". Inclua a verificação de timestamp desde o primeiro dia, deixe o histórico fora do hash e deixe o MQTT continuar fazendo o que faz de melhor: transportar as atualizações.

Perguntas frequentes