Store and Forward no MQTT: Buffer de Dados na Borda Durante Quedas e Reenvio em Ordem

Store and forward MQTT é o padrão que mantém os dados industriais intactos quando a rede entre a borda e a nuvem cai. Em vez de descartar mensagens enquanto o broker está inacessível, o dispositivo de borda as grava em armazenamento local e as reenvia, em ordem, quando a conexão volta. Este artigo explica como o padrão funciona, quais decisões de projeto importam e traz uma implementação em Python pronta para rodar, baseada em SQLite.

Por Que Só o MQTT Não Basta

O MQTT tem recursos fortes de entrega: confirmações de QoS 1 e 2, sessões persistentes e retained messages. Mas todos eles só começam a funcionar depois que a mensagem saiu do dispositivo. Veja o que acontece durante uma queda de WAN de duas horas em uma planta, por exemplo uma mineradora no interior ou um frigorífico com um único link:

• O gateway continua lendo as máquinas e gerando valores.
• O cliente MQTT não consegue se conectar, então não há sessão para onde entregar.
• A maioria das bibliotecas cliente guarda mensagens não enviadas em memória, com limite de tamanho, e as perde em um reinício ou falta de energia.

Sessões persistentes no broker ajudam na direção oposta: guardam mensagens para um assinante offline. Elas não fazem nada por um publicador que não consegue alcançar o broker. É essa lacuna que o store and forward preenche, e é por isso que o buffer aparece como função essencial de todo gateway de borda IIoT.

Como Funciona o Store and Forward no MQTT

  Data source --> [ write to local queue (disk) ] --> forwarder --> MQTT broker
                         ^                               |
                         |      delete only after        |
                         +------- PUBACK received -------+

1. Toda mensagem é gravada primeiro em uma fila local durável, com um número de sequência e o timestamp de origem.
2. Um encaminhador lê as mensagens mais antigas e as publica com QoS 1.
3. Somente depois que o broker confirma (PUBACK) a mensagem é apagada da fila.
4. Quando a conexão falha, o encaminhador para, a fila cresce e as fontes de dados seguem sem impacto.
5. Na reconexão, o encaminhador esvazia o acúmulo da mais antiga para a mais nova, a uma taxa controlada, e depois continua com os dados ao vivo.

Fazer todos os dados passarem pela fila, e não apenas os gerados durante as quedas, mantém a ordenação simples: há um único caminho, e ele é sempre first-in, first-out.

Decisões de Projeto Que Importam

Dimensionando o Buffer

Estime-o a partir de três números: mensagens por segundo em operação normal, tamanho médio da mensagem incluindo overhead e a maior queda que você quer suportar. Por exemplo, 20 mensagens por segundo de 200 bytes durante uma queda de 24 horas somam cerca de 345 MB de payload, mais o overhead de armazenamento. Se for demais, reduzir o tráfego na origem com report by exception costuma diminuir drasticamente a necessidade.

Escolhendo uma Política de Descarte

Quando o buffer enche, é preciso escolher: descartar os dados mais antigos (manter o estado recente) ou os mais novos (manter um histórico contínuo desde o início da queda). Para monitoramento, descartar os mais antigos geralmente é o certo. Para conformidade ou registros de lote, você pode preferir parar de aceitar novos dados de baixa prioridade e preservar o início. De qualquer forma, publique um contador de mensagens descartadas para que a lacuna fique visível.

Reenvio e o Último Valor Retido

Reenviar horas de histórico para um topic que os dashboards também usam como "valor atual" pode fazer um mostrador saltar por valores antigos. Duas soluções comuns são enviar o acúmulo para um topic separado, como .../history, ou incluir o timestamp de origem e deixar os consumidores ignorarem valores mais antigos na exibição. Com o MQTT 5, você também pode definir um message expiry interval para valores que perdem a utilidade quando ficam velhos, para que o broker os descarte em vez de entregá-los atrasados.

Uma Implementação de Store and Forward em Python

O script a seguir usa SQLite como fila durável e paho-mqtt 2.x como cliente. Uma thread produtora gera uma leitura por segundo, independentemente da conectividade. O encaminhador publica as linhas mais antigas com QoS 1, espera cada confirmação, apaga as linhas entregues e limita o reenvio para ficar abaixo do limite de 10 mensagens por segundo do servidor MQTT público.

import json
import sqlite3
import threading
import time
import random
from datetime import datetime, timezone

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
CLIENT_ID = "saf-gateway-2b7c"
TOPIC = "connectmq/demo/saf-2b7c/plant1/compressor1/telemetry"
DB_PATH = "saf_queue.db"
MAX_ROWS = 500_000          # limite do buffer; as linhas mais antigas são descartadas além disso
REPLAY_RATE = 8             # mensagens por segundo (o broker público permite 10)
ACK_TIMEOUT = 10            # segundos de espera pelo PUBACK

db_lock = threading.Lock()
db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("""CREATE TABLE IF NOT EXISTS queue (
                seq INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload TEXT NOT NULL)""")
db.commit()


def enqueue(topic, data):
    with db_lock:
        db.execute("INSERT INTO queue (topic, payload) VALUES (?, ?)",
                   (topic, json.dumps(data)))
        count = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
        if count > MAX_ROWS:
            db.execute("DELETE FROM queue WHERE seq IN "
                       "(SELECT seq FROM queue ORDER BY seq LIMIT ?)", (count - MAX_ROWS,))
        db.commit()


def oldest(n=50):
    with db_lock:
        return db.execute("SELECT seq, topic, payload FROM queue ORDER BY seq LIMIT ?",
                          (n,)).fetchall()


def delete(seq):
    with db_lock:
        db.execute("DELETE FROM queue WHERE seq = ?", (seq,))
        db.commit()


def producer():
    """Fonte de dados simulada: nunca bloqueia esperando a rede."""
    while True:
        enqueue(TOPIC, {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "pressure_bar": round(7 + random.gauss(0, 0.1), 3),
            "motor_temp_c": round(65 + random.gauss(0, 0.5), 2),
        })
        time.sleep(1)


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("Disconnected:", reason_code, "- buffering locally")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
# Produção: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); PORT = 8883
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

threading.Thread(target=producer, daemon=True).start()

interval = 1.0 / REPLAY_RATE
while True:
    if not client.is_connected():
        time.sleep(1)
        continue
    rows = oldest()
    if not rows:
        time.sleep(0.5)
        continue
    for seq, topic, payload in rows:
        data = json.loads(payload)
        data["seq"] = seq                      # permite aos consumidores eliminar duplicatas
        try:
            info = client.publish(topic, json.dumps(data), qos=1)
            info.wait_for_publish(timeout=ACK_TIMEOUT)
        except (RuntimeError, ValueError) as exc:
            print("Publish failed:", exc)
            break
        if not info.is_published():
            print("No PUBACK, will retry from seq", seq)
            break                              # mantém a linha e tenta de novo depois, em ordem
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

Para testar, rode o script, acompanhe o topic em outro terminal e depois desconecte a sua rede por um ou dois minutos. O exemplo usa o servidor MQTT público, adequado só para testes e dados simulados:

mosquitto_sub -h broker.connectmq.com -p 1883 -i saf-2b7c-viewer -t "connectmq/demo/saf-2b7c/#" -v

Enquanto está offline, o script imprime a desconexão e continua enfileirando. Quando a rede volta, o cliente reconecta automaticamente e o acúmulo é esvaziado na ordem de sequência. O campo ts mostra quando cada valor foi medido, e seq cresce sem lacunas. Você também pode parar o script completamente durante a queda; ao reiniciar, a fila no SQLite continua lá.

Observações sobre a Implementação

• Uma mensagem em trânsito por vez mantém a ordenação trivial e é rápido o suficiente para a maioria dos gateways. Para maior vazão, publique uma pequena janela e apague as linhas à medida que as confirmações chegam.
• Pelo menos uma vez, não exatamente uma vez. Se o PUBACK se perder depois que o broker aceitou a mensagem, a linha é reenviada. O campo seq permite que um historiador ou assinante descarte duplicatas. O QoS 2 do MQTT evitaria duplicatas na rede, mas não através de um reinício do gateway, então IDs no nível da aplicação continuam valendo a pena. Veja níveis de QoS do MQTT explicados para as contrapartidas.
• O modo WAL do SQLite permite que leituras e escritas aconteçam ao mesmo tempo, com boa durabilidade.
• O reenvio com taxa limitada protege o broker e os consumidores. Planos compartilhados têm limites por cliente, e o plano Serverless gratuito da ConnectMQ suporta QoS 1, que é exatamente o que este padrão precisa.

Checklist de Store and Forward MQTT

1. Grave toda mensagem em disco antes de tentar enviá-la.
2. Inclua um timestamp de origem e um número de sequência em todo payload.
3. Apague do buffer apenas depois que o broker confirmar.
4. Reenvie da mais antiga para a mais nova, com taxa abaixo dos limites do broker e dos consumidores.
5. Defina um tamanho máximo de buffer e uma política de descarte, e reporte os descartes.
6. Decida como o histórico reenviado interage com os topics de "valor atual".
7. Teste com quedas reais: puxe o cabo, reinicie o gateway no meio da queda, encha o buffer.

Store and Forward em Rede Celular e Sites Remotos

Ativos remotos em links celulares perdem conectividade com mais frequência e pagam por byte. Nesses casos, o store and forward combina naturalmente com agrupamento (várias leituras por mensagem) e compressão. O guia de MQTT sobre rede celular aborda o ajuste do keep alive e o orçamento de dados desses links, e o guia de IoT industrial com MQTT mostra onde o buffer se encaixa em uma arquitetura completa de planta.

Um Broker Confiável do Outro Lado

O store and forward protege o lado da borda da conexão. No lado da nuvem, você precisa de um broker acessível, protegido com TLS e dimensionado para rajadas de reenvio. Um broker MQTT gerenciado tira a operação do broker das suas mãos, e o guia de broker MQTT na nuvem explica como comparar as opções.

Comece a Usar a ConnectMQ

Aponte seus gateways com buffer para a ConnectMQ: TLS na porta 8883, credenciais emitidas, message expiry do MQTT 5 e um explorador de topics ao vivo para ver o acúmulo sendo esvaziado após uma queda. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito.

Perguntas frequentes

O MQTT tem store and forward embutido?

Não para publicadores que não conseguem alcançar o broker. QoS e sessões persistentes funcionam depois que a mensagem chega ao broker; o buffer antes disso é tarefa da aplicação de borda.

Devo fazer o buffer em memória ou em disco?

Em disco. Buffers em memória se perdem em um reinício ou falta de energia, o que muitas vezes acontece justamente nos mesmos eventos que derrubam a rede.

Como os consumidores lidam com mensagens duplicadas após o reenvio?

Inclua um número de sequência ou ID único em cada payload e deixe os consumidores ignorarem IDs que já armazenaram. Historiadores muitas vezes também eliminam duplicatas por timestamp e tag.

Um reenvio grande vai sobrecarregar o broker?

Pode estourar limites de taxa ou sobrecarregar os assinantes se for enviado de uma vez. Limite o reenvio a uma taxa fixa, como no exemplo, e dimensione a taxa de acordo com o seu plano.