MQTT and Redis: Cache the Latest Device State

To cache the latest device state from MQTT in Redis, run a small subscriber that listens on your telemetry topics and writes each message into a Redis hash keyed by device ID. Your APIs, dashboards and background jobs then read the current state with a single HGETALL, without opening an MQTT connection or waiting for the next message.

This tutorial builds that subscriber in Python, protects it against out-of-order updates, adds a Redis Stream for recent history, and compares the approach with retained messages and Redis Pub/Sub.

When a Redis cache makes sense

MQTT already has a "latest value" feature: retained messages. A new subscriber immediately receives the last retained message on each topic. So why add Redis?

If only MQTT clients need the last value, retained messages are enough. As soon as non-MQTT code needs state, a Redis cache is simpler and faster.

The data model

Keep it boring and predictable:

• device:<id> — a hash with the latest fields: temp, humidity, battery, ts (device timestamp), topic, seen (when the bridge wrote it).
• devices:online — an optional set of device IDs, maintained from status topics.
• telemetry — an optional stream with recent raw messages for replay and debugging.

Devices publish JSON to testmqtt/redis/<device>/state, for example {"ts": 1760700000123, "temp": 21.4, "battery": 87}. A timestamp from the device (or a sequence number) is what lets you reject stale updates later. See JSON payloads in Python for parsing tips.

The subscriber

This bridge uses paho-mqtt 2.x and redis-py. The Lua script writes the hash only if the incoming timestamp is newer than the stored one, and refreshes a TTL so silent devices disappear by themselves:

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

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

# KEYS[1] = device key; ARGV = ts, ttl, then field/value pairs
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   # forget devices that are silent for 15 minutes

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", 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("skipping non-JSON payload on", 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/objects as 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"stale update for {device} ignored (ts={ts})")

    # optional short history, capped at ~100k entries
    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()

Notes on the design:

• Atomic compare-and-set. Lua scripts run atomically in Redis, so two bridge instances (or a QoS 1 redelivery) cannot race each other into writing an older state.
• Flat fields. Strings and numbers go in as-is; booleans, nulls and nested objects are stored as JSON strings. Redis hashes are one level deep — do not fight that.
• TTL as an offline signal. If a key expires, the device has not reported for TTL_SECONDS. For an explicit signal, combine it with a Last Will message on a status topic.
• Duplicates are harmless. Writing the same state twice changes nothing, which makes this sink naturally idempotent.

Reading the state

Any service can now answer "what is device X doing right now?" without 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:*'

For a dashboard that needs live updates, keep the REST endpoint backed by Redis for the initial page load, and let the browser subscribe to MQTT over secure WebSockets (port 8084, path /mqtt) for changes after that. The cache answers "now", MQTT delivers "next".

Redis Streams for recent history

The XADD line keeps a capped log of raw messages. Streams support consumer groups, so several workers can process updates with acknowledgements, and you can replay a time range when debugging:

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

Keep the stream short (MAXLEN) — it lives in memory. For long-term history write to a time-series database or Postgres instead; see MQTT to PostgreSQL and MQTT data retention and storage.

MQTT vs Redis Pub/Sub

Redis also has a Pub/Sub feature, and it is tempting to skip MQTT entirely. The two are not interchangeable:

• Redis Pub/Sub is fire-and-forget between processes that are connected to Redis right now. There is no QoS, no retained value, no Last Will, no per-topic ACL model designed for devices, and clients must hold a Redis connection.
• MQTT is designed for devices on unreliable networks: QoS 1 and 2, persistent sessions, retained messages, keep alive, Last Will and lightweight clients for microcontrollers.

A common split is MQTT at the edge and Redis inside the backend. The longer comparison is in MQTT vs Redis Pub/Sub.

Test it

Start Redis locally and run the bridge:

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

Publish a fresh state, then an older one with a smaller timestamp:

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 must still be 22.5, and the bridge should log stale update for sensor-1 ignored. You can also send the payloads by hand from the online MQTT client and watch redis-cli MONITOR to see each command the bridge issues.

Publish test states from your browser

A Redis state cache is one of the cheapest upgrades to an MQTT system: one subscriber, one hash per device, and every service gets instant access to "now". Add the timestamp check from day one, keep history out of the hash, and let MQTT keep doing what it does best — moving the updates.

Frequently asked questions