Buffer de mensajes MQTT offline en Python con SQLite (store and forward)

Para crear un buffer de mensajes MQTT offline en Python, escribe primero cada lectura en una cola SQLite local y deja que un emisor independiente vacíe esa cola hacia el broker solo cuando haya conexión. En este tutorial construirás ese publicador store-and-forward con paho-mqtt 2.x: sobrevive a cortes de red y reinicios, reenvía los mensajes en el orden original, los borra solo después de que el broker los confirme y mantiene acotado el uso de disco. Funciona en una Raspberry Pi o en cualquier máquina Linux.

Por qué no basta la cola integrada de paho-mqtt

paho-mqtt ya encola los mensajes salientes mientras se reconecta, así que ¿para qué construir nada? Porque esa cola vive en memoria. Si el proceso se reinicia, la Pi rearranca o se corta la alimentación durante un corte de red, todo lo que contiene se pierde. En un gateway con conexión celular, en un sitio con Wi-Fi poco fiable o en un vehículo que atraviesa túneles, esos son justo los momentos que importan.

También hay un problema sutil en cómo trata el broker a los publicadores. Las sesiones persistentes y QoS 1 o 2 protegen los mensajes hacia suscriptores que están temporalmente desconectados. No hacen nada por un publicador que no puede llegar al broker. La durabilidad en el lado del publicador es tarea del cliente.

El diseño que resuelve ambos problemas es un outbox clásico:

1. La aplicación nunca publica directamente. Inserta una fila en una tabla SQLite: topic, payload, QoS, indicador retain y marca de tiempo.
2. Un hilo emisor lee las filas más antiguas, las publica con QoS 1 y espera el PUBACK del broker.
3. Solo se borran las filas confirmadas. Todo lo demás se queda para el siguiente intento.

Como SQLite es un archivo, la cola sobrevive a los reinicios. Como el emisor procesa las filas en orden de inserción sobre una única conexión, el orden se conserva.

Diseño de la tabla outbox

outbox
  id        INTEGER PRIMARY KEY AUTOINCREMENT   -- defines replay order
  topic     TEXT NOT NULL
  payload   BLOB NOT NULL
  qos       INTEGER NOT NULL DEFAULT 1
  retain    INTEGER NOT NULL DEFAULT 0
  created   REAL NOT NULL                      -- unix time of the reading

Dos ajustes de SQLite marcan una gran diferencia en una Pi. journal_mode=WAL permite que el productor inserte mientras el emisor lee y reduce el número de escrituras por transacción. synchronous=NORMAL es un equilibrio razonable entre seguridad y desgaste de la tarjeta SD en modo WAL. Coloca la base de datos en un SSD o en una partición escribible si usas un sistema de archivos raíz de solo lectura.

El publicador store-and-forward

El módulo siguiente es autocontenido. Importa OfflineBufferedPublisher, llama a publish() desde cualquier punto de tu aplicación y él se encarga del resto.

import json
import os
import sqlite3
import threading
import time
import uuid

import paho.mqtt.client as mqtt


class OfflineBufferedPublisher:
    def __init__(self, host, port=1883, client_id=None, username=None, password=None,
                 db_path="outbox.db", max_rows=200_000, batch=20, max_rate=8.0):
        self.db_path = db_path
        self.max_rows = max_rows
        self.batch = batch
        self.min_interval = 1.0 / max_rate if max_rate else 0
        self._lock = threading.Lock()
        self._wake = threading.Event()
        self._stop = threading.Event()

        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=NORMAL")
        self.db.execute("""CREATE TABLE IF NOT EXISTS outbox (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL, payload BLOB NOT NULL,
            qos INTEGER NOT NULL DEFAULT 1, retain INTEGER NOT NULL DEFAULT 0,
            created REAL NOT NULL)""")
        self.db.commit()

        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=client_id or f"buffered-{uuid.uuid4().hex[:8]}")
        if username:
            self.client.username_pw_set(username, password)
        if port == 8883:
            self.client.tls_set()
        self.client.reconnect_delay_set(1, 60)
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.connect_async(host, port, keepalive=30)
        self.client.loop_start()

        self.sender = threading.Thread(target=self._drain, daemon=True)
        self.sender.start()

    # ---- producer side -------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if not isinstance(payload, (bytes, bytearray)):
            payload = json.dumps(payload).encode() if not isinstance(payload, str) else payload.encode()
        with self._lock:
            self.db.execute("INSERT INTO outbox (topic, payload, qos, retain, created) VALUES (?,?,?,?,?)",
                            (topic, payload, qos, int(retain), time.time()))
            self._enforce_limit()
            self.db.commit()
        self._wake.set()

    def pending(self):
        with self._lock:
            return self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]

    def _enforce_limit(self):
        count = self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
        if count > self.max_rows:
            self.db.execute("DELETE FROM outbox WHERE id IN (SELECT id FROM outbox ORDER BY id LIMIT ?)",
                            (count - self.max_rows,))

    # ---- sender side ---------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        print("connected:", reason_code)
        self._wake.set()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("disconnected:", reason_code)

    def _drain(self):
        while not self._stop.is_set():
            if not self.client.is_connected():
                self._wake.wait(2)
                self._wake.clear()
                continue
            with self._lock:
                rows = self.db.execute(
                    "SELECT id, topic, payload, qos, retain FROM outbox ORDER BY id LIMIT ?",
                    (self.batch,)).fetchall()
            if not rows:
                self._wake.wait(5)
                self._wake.clear()
                continue
            for row_id, topic, payload, qos, retain in rows:
                info = self.client.publish(topic, payload, qos=qos, retain=bool(retain))
                try:
                    info.wait_for_publish(timeout=10)
                except (ValueError, RuntimeError):
                    break
                if not info.is_published():
                    break  # connection dropped; keep this row and everything after it
                with self._lock:
                    self.db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
                    self.db.commit()
                if self.min_interval:
                    time.sleep(self.min_interval)

    def close(self):
        self._stop.set()
        self._wake.set()
        self.sender.join(timeout=5)
        self.client.loop_stop()
        self.client.disconnect()
        self.db.close()


if __name__ == "__main__":
    pub = OfflineBufferedPublisher(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        client_id=os.getenv("MQTT_CLIENT_ID"),
        username=os.getenv("MQTT_USERNAME"),
        password=os.getenv("MQTT_PASSWORD"),
    )
    prefix = os.getenv("TOPIC_PREFIX", "connectmq/demo/q5w8e1/sensor1")
    n = 0
    try:
        while True:
            n += 1
            pub.publish(f"{prefix}/reading", {"seq": n, "ts": round(time.time(), 3), "value": 20 + n % 5})
            print("queued", n, "pending", pub.pending())
            time.sleep(2)
    except KeyboardInterrupt:
        pub.close()

Probar un corte de red

Inicia un suscriptor que imprima los números de secuencia y después ejecuta el publicador:

mosquitto_sub -h broker.connectmq.com -p 1883 -i buffer-viewer-q5w8e1 -t "connectmq/demo/q5w8e1/#" -v
python buffered_publisher.py

Ahora simula un corte. En una Pi puedes desactivar la red con sudo ip link set wlan0 down (hazlo desde una consola local, no por SSH sobre la misma interfaz), esperar un minuto y volver a activarla con sudo ip link set wlan0 up. El contador "pending" crece durante el corte. Cuando paho se reconecta, el emisor vacía la cola acumulada y el suscriptor recibe en orden todos los números de secuencia que faltaban, seguidos de las lecturas en directo. Prueba a detener el script durante el corte y a iniciarlo de nuevo: la cola acumulada sigue ahí.

Orden, duplicados y marcas de tiempo

Orden. El emisor publica un mensaje cada vez y espera su confirmación antes de borrarlo y pasar al siguiente. Sobre una única conexión MQTT, eso conserva el orden original. Puedes aumentar el rendimiento permitiendo varios mensajes en vuelo, pero mantén un único hilo emisor y una única conexión si el orden importa.

Duplicados. QoS 1 garantiza al menos una entrega. Si la conexión cae después de que el broker reciba un mensaje pero antes de que llegue el PUBACK, la fila se envía de nuevo tras la reconexión. Incluye un número de secuencia o un ID único en el payload para que los consumidores puedan descartar duplicados. Los niveles de QoS en MQTT explica los compromisos y por qué QoS 2 tampoco te protege de un reinicio del cliente.

Marcas de tiempo. Un mensaje reenviado llega minutos u horas tarde. Pon siempre la hora de la medición en el payload y haz que los consumidores la usen en lugar de la hora de llegada. Con MQTT 5 también puedes fijar un intervalo de expiración del mensaje en lecturas que pierden sentido pasado un tiempo, como una posición en directo.

Ajustes para despliegues reales

Cuando la cola está llena, esta implementación descarta las filas más antiguas, lo que encaja con la telemetría, donde lo que más importa son los datos recientes. Para registros de eventos quizá prefieras dejar de aceptar filas nuevas y lanzar una alarma. En cualquier caso, publica el valor de pending() como métrica de salud para que una cola creciente sea visible en tu dashboard.

Ejecuta el publicador como servicio systemd para que se reinicie con el dispositivo, como se muestra en ejecutar un script MQTT de Python al arrancar. Para otras formas de almacenar datos MQTT en local, consulta MQTT a SQLite en Python, que cubre el lado del suscriptor.

Alternativa: un broker local con un bridge

Otro diseño store-and-forward habitual coloca un broker Mosquitto en la propia Pi. Las aplicaciones locales publican en localhost y Mosquitto hace de bridge de los topics hacia el broker en la nube, encolando los mensajes mientras el enlace está caído. Encaja bien cuando varios procesos del mismo dispositivo publican datos o cuando no puedes cambiar el código de la aplicación. La guía de configuración del bridge de Mosquitto cubre los ajustes.

Conviene conocer los compromisos. Ahora operas un broker en cada gateway, con su propia configuración, actualizaciones y credenciales. Los límites de cola y el comportamiento de persistencia de Mosquitto requieren un ajuste cuidadoso para sobrevivir a los reinicios, y depurar huecos en los datos implica revisar dos brokers en lugar de uno. El outbox en Python mantiene todo dentro de un único proceso que controlas por completo y que puedes probar con un test unitario, algo que muchos equipos consideran más sencillo para un gateway de propósito único. Elige el bridge cuando muchos productores locales compartan un mismo enlace, y el outbox cuando una sola aplicación sea dueña de los datos.

Ajustes del broker en producción

El servidor MQTT público gratuito sirve para esta prueba, pero todo lo que se publica allí es público; no lo uses para datos de producción. Para datos reales, crea un broker privado en ConnectMQ y arranca el publicador con MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD y MQTT_CLIENT_ID=YOUR_CLIENT_ID obtenidos del portal. Un broker MQTT gestionado se ocupa de la disponibilidad en el lado del servidor, mientras que este outbox cubre los huecos en el lado del cliente. La documentación de paho-mqtt describe en detalle MQTTMessageInfo y las opciones de reconexión.

Este patrón es una pieza básica de cada gateway de la serie tutorial de Raspberry Pi con MQTT, desde lectores de bus CAN hasta receptores LoRa.

No pierdas ni una lectura con ConnectMQ

Combina un outbox local duradero con un broker fiable y los cortes de red se vuelven invisibles para tus dashboards. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, apunta este publicador hacia él por TLS y observa cómo la cola acumulada se vacía en el explorador de topics en directo de ConnectMQ tras tu próximo corte de red.

Preguntas frecuentes

¿La sesión persistente de MQTT guarda en buffer los mensajes de un publicador?

No. Las sesiones persistentes guardan mensajes para suscriptores desconectados. Un publicador que no puede llegar al broker debe almacenar los mensajes en local, que es lo que hace este outbox.

¿Llegarán los mensajes reenviados en el orden correcto?

Sí, siempre que un único emisor publique las filas en orden de ID sobre una única conexión y espere cada confirmación, como en este código.

¿Puedo usarlo con asyncio?

Sí. El mismo diseño de outbox funciona con aiomqtt: ejecuta el bucle de vaciado como tarea asíncrona y usa aiosqlite o ejecuta las llamadas a SQLite en un executor de hilos.

¿Cómo evito desgastar la tarjeta SD?

Usa el modo WAL, mantén los payloads compactos y coloca la base de datos en un SSD siempre que sea posible. Escribir una lectura cada pocos segundos es una carga modesta para un almacenamiento de calidad.