Buffer de Mensagens MQTT Offline em Python com SQLite (Store and Forward)

Para fazer buffer de mensagens MQTT offline em Python, grave cada leitura primeiro em uma fila SQLite local e deixe um remetente separado esvaziar essa fila para o broker apenas quando a conexão estiver ativa. Neste tutorial você vai construir esse publicador store-and-forward com paho-mqtt 2.x: ele sobrevive a quedas de rede e reinicializações, reenvia as mensagens na ordem original, só as apaga depois que o broker as confirma e mantém o uso de disco limitado. Ele roda em um Raspberry Pi ou em qualquer máquina Linux.

Por que a fila interna do paho-mqtt não basta

O paho-mqtt já enfileira mensagens de saída enquanto reconecta, então por que construir algo? Porque essa fila vive na memória. Se o processo reiniciar, o Pi reiniciar ou a energia cair durante uma queda de rede, tudo o que está nela se perde. Para um gateway em rede celular, um site com Wi-Fi instável ou um veículo que passa por túneis, esses são exatamente os momentos que importam.

Há também um problema sutil na forma como o broker trata os publicadores. Sessões persistentes e QoS 1 ou 2 protegem mensagens para assinantes temporariamente offline. Elas não fazem nada por um publicador que não consegue alcançar o broker. A durabilidade do lado da publicação é responsabilidade do client.

O design que resolve os dois problemas é uma clássica outbox (caixa de saída):

1. A aplicação nunca publica diretamente. Ela insere uma linha em uma tabela SQLite: topic, payload, QoS, flag de retenção e timestamp.
2. Uma thread remetente lê as linhas mais antigas, publica-as em QoS 1 e aguarda o PUBACK do broker.
3. Apenas linhas confirmadas são apagadas. Todo o resto fica para a próxima tentativa.

Como o SQLite é um arquivo, a fila sobrevive a reinicializações. Como o remetente processa as linhas na ordem de inserção por uma única conexão, a ordem é preservada.

Projetando a tabela 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

Duas configurações do SQLite fazem grande diferença em um Pi. journal_mode=WAL permite que o produtor insira enquanto o remetente lê e reduz o número de gravações por transação. synchronous=NORMAL é um equilíbrio razoável entre segurança e desgaste do cartão SD no modo WAL. Coloque o banco em um SSD ou em uma partição gravável se você usa um sistema de arquivos raiz somente leitura.

O publicador store-and-forward

O módulo abaixo é autocontido. Importe OfflineBufferedPublisher, chame publish() de qualquer lugar da sua aplicação e ele cuida do 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()

    # ---- lado do produtor ----------------------------------------------
    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,))

    # ---- lado do remetente ---------------------------------------------
    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  # a conexão caiu; mantém esta linha e todas as seguintes
                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()

Testando uma queda de rede

Inicie um assinante que imprime os números de sequência e depois execute o publicador:

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

Agora simule uma queda. Em um Pi você pode derrubar a rede com sudo ip link set wlan0 down (faça isso de um console local, não por SSH na mesma interface), esperar um minuto e trazê-la de volta com sudo ip link set wlan0 up. O contador "pending" cresce durante a queda. Quando o paho reconecta, o remetente esvazia o acúmulo e o assinante recebe, em ordem, todos os números de sequência que faltavam, seguidos das leituras ao vivo. Experimente parar o script durante a queda e iniciá-lo novamente: o acúmulo continua lá.

Ordem, duplicatas e timestamps

Ordem. O remetente publica uma mensagem por vez e aguarda sua confirmação antes de apagá-la e seguir em frente. Em uma única conexão MQTT, isso preserva a ordem original. Você pode aumentar a vazão permitindo várias mensagens em trânsito, mas mantenha uma única thread remetente e uma única conexão se a ordem importar.

Duplicatas. QoS 1 é entrega "pelo menos uma vez". Se a conexão cair depois que o broker recebeu uma mensagem, mas antes de o PUBACK chegar, a linha é enviada novamente após a reconexão. Inclua um número de sequência ou um ID único no payload para que os consumidores possam descartar duplicatas. Níveis de QoS do MQTT explica os trade-offs e por que nem o QoS 2 protege você contra uma reinicialização do client.

Timestamps. Uma mensagem reenviada chega minutos ou horas atrasada. Sempre coloque o horário da medição no payload e faça os consumidores usarem esse valor, e não o horário de chegada. Com MQTT 5 você também pode definir um intervalo de expiração de mensagem para leituras que perdem o sentido depois de um tempo, como uma posição ao vivo.

Ajustes para implantações reais

Quando a fila está cheia, esta implementação descarta as linhas mais antigas, o que se adequa a telemetria em que os dados recentes importam mais. Para logs de eventos, talvez você prefira parar de aceitar novas linhas e disparar um alarme. De qualquer forma, publique a contagem de pending() como métrica de saúde, para que um acúmulo crescente fique visível no seu dashboard.

Execute o publicador como serviço do systemd para que ele reinicie junto com o dispositivo, como mostrado em executar um script Python MQTT na inicialização. Para outras formas de armazenar dados MQTT localmente, veja MQTT para SQLite em Python, que cobre o lado do assinante.

Alternativa: um broker local com bridge

Outro design store-and-forward comum coloca um broker Mosquitto no próprio Pi. As aplicações locais publicam em localhost, e o Mosquitto faz a bridge dos topics para o broker na nuvem, enfileirando mensagens enquanto o uplink está fora. É uma boa opção quando vários processos no mesmo dispositivo publicam dados, ou quando você não pode alterar o código da aplicação. O guia de configuração de bridge do Mosquitto cobre as configurações.

Vale conhecer os trade-offs. Agora você opera um broker em cada gateway, com configuração, atualizações e credenciais próprias. Os limites de fila e o comportamento de persistência do Mosquitto exigem ajuste cuidadoso para sobreviver a reinicializações, e depurar lacunas de dados significa olhar para dois brokers em vez de um. A outbox em Python mantém tudo dentro de um único processo que você controla totalmente e pode testar com um teste unitário, o que muitas equipes acham mais simples para um gateway de finalidade única. Escolha a bridge quando muitos produtores locais compartilham um uplink, e a outbox quando uma única aplicação é dona dos dados.

Configurações de broker para produção

O servidor MQTT público gratuito serve para este teste, mas tudo o que é publicado ali é público; ele é para testes e protótipos, não para dados de produção. Para dados reais, crie um broker privado na ConnectMQ e inicie o publicador com MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD e MQTT_CLIENT_ID=YOUR_CLIENT_ID obtidos no portal. Um broker MQTT gerenciado cuida da disponibilidade do lado do servidor, enquanto esta outbox cobre as lacunas do lado do client. A documentação do paho-mqtt descreve em detalhes o MQTTMessageInfo e as opções de reconexão.

Esse padrão é um bloco fundamental para todo gateway da série tutorial de MQTT com Raspberry Pi, de leitores de CAN bus a receptores LoRa.

Nunca perca uma leitura com a ConnectMQ

Combine uma outbox local durável com um broker confiável e as quedas de rede ficam invisíveis para os seus dashboards. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, aponte este publicador para ele via TLS e veja o acúmulo escoar para o explorador de topics ao vivo da ConnectMQ depois da sua próxima queda de rede.

Perguntas frequentes

A sessão persistente do MQTT faz buffer de mensagens para um publicador?

Não. Sessões persistentes guardam mensagens para assinantes offline. Um publicador que não consegue alcançar o broker precisa fazer buffer localmente, que é o que esta outbox faz.

As mensagens reenviadas chegam na ordem certa?

Sim, desde que um único remetente publique as linhas em ordem de ID por uma única conexão e aguarde cada confirmação, como neste código.

Posso usar isso com asyncio?

Sim. O mesmo design de outbox funciona com aiomqtt: execute o loop de esvaziamento como uma tarefa assíncrona e use aiosqlite ou execute as chamadas SQLite em um executor de threads.

Como evito desgastar o cartão SD?

Use o modo WAL, mantenha os payloads compactos e coloque o banco em um SSD sempre que possível. Gravar uma leitura a cada poucos segundos é uma carga modesta para um armazenamento de boa qualidade.