Reconexão Paho MQTT: Backoff Robusto, Reassinatura e Buffer Offline em Python

O comportamento de reconexão do Paho MQTT é a diferença entre um cliente que sobrevive a uma rede instável e outro que para de funcionar em silêncio às 3 da manhã. Este guia explica o que o paho-mqtt 2.x faz automaticamente, as três lacunas que ele deixa abertas, e entrega um cliente Python testado com backoff exponencial, reassinatura no connect, um buffer offline e sessões MQTT 5 persistentes.

Sintomas de uma lógica de reconexão quebrada

Bugs de reconexão raramente derrubam o programa. Eles aparecem como falhas mais discretas, fáceis de diagnosticar errado:

• O cliente reconecta depois de uma queda do Wi-Fi, registra "connected", mas nunca mais recebe uma mensagem.
• Um script iniciado com a rede fora do ar termina na hora com um erro de socket, em vez de esperar.
• Leituras feitas durante uma queda curta desaparecem, deixando buracos nos gráficos.
• Centenas de dispositivos reconectam exatamente no mesmo instante depois de um reinício do broker e sobrecarregam a rede.
• Dois processos ficam se desconectando a cada poucos segundos.

Cada um desses sintomas tem uma causa específica e uma correção específica.

O que o Paho faz automaticamente

Quando você executa o network loop com loop_start() ou loop_forever(), o paho-mqtt reconecta sozinho depois que uma conexão estabelecida cai. O intervalo entre as tentativas começa em min_delay e dobra a cada falha até chegar a max_delay, configurados com reconnect_delay_set(). Os valores padrão são 1 e 120 segundos. Uma conexão bem-sucedida reinicia o intervalo.

O paho também guarda as mensagens QoS 1 e QoS 2 em trânsito que o broker ainda não confirmou e as retransmite depois da reconexão, desde que a sessão seja retomada.

Isso cobre muita coisa, mas não tudo. As lacunas restantes são responsabilidade sua.

Lacuna 1: a primeira conexão falha

client.connect() faz a primeira conexão de forma síncrona e lança uma exceção, normalmente ConnectionRefusedError, socket.gaierror ou TimeoutError, se o broker estiver inacessível. A reconexão automática nunca começa, porque nunca houve uma conexão. É por isso que scripts iniciados no boot, antes de a rede estar pronta, costumam morrer.

A correção é connect_async() seguido de loop_start(). A tentativa de conexão passa para o network loop, que tenta de novo usando as mesmas regras de backoff.

Lacuna 2: as assinaturas desaparecem

Com uma sessão limpa, o broker descarta as suas assinaturas quando a conexão termina. O paho não as registra e não reassina por você. Se você chamar subscribe() uma única vez depois de connect(), o cliente vai reconectar feliz da vida e depois não vai receber nada.

A correção é simples e inegociável: assine dentro do on_connect. Esse callback roda depois de cada conexão bem-sucedida, inclusive nas reconexões, então as assinaturas são sempre restauradas.

Lacuna 3: mensagens publicadas enquanto offline

As leituras produzidas enquanto o link está fora do ar precisam ir para algum lugar. Mensagens QoS 0 são "dispare e esqueça", então publicá-las desconectado significa perdê-las. Depender da fila interna do paho para níveis de QoS maiores também não é um bom plano, porque ela é ilimitada e invisível para a sua aplicação. Um buffer offline explícito e limitado, esvaziado depois da reconexão, dá a você controle sobre a memória e sobre o que descartar quando uma queda dura demais.

Um cliente robusto de reconexão Paho MQTT

A classe abaixo fecha as três lacunas e adiciona jitter ao backoff, para que uma frota de dispositivos não reconecte em sincronia. Ela usa MQTT 5 com session expiry, para que o broker mantenha as assinaturas e as mensagens QoS 1 enfileiradas durante quedas curtas.

pip install "paho-mqtt>=2.0"

import json
import logging
import random
import threading
import time
import uuid
from collections import deque

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("mqtt")


class ResilientClient:
    def __init__(self, host, port, client_id, subscriptions,
                 username=None, password=None, use_tls=False,
                 session_expiry_s=1800, buffer_size=1000):
        self.host = host
        self.port = port
        self.subscriptions = subscriptions  # lista de (topic, qos)
        self.session_expiry_s = session_expiry_s
        self.offline = deque(maxlen=buffer_size)  # a mais antiga é descartada quando cheio
        self.lock = threading.Lock()
        self.on_payload = None

        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                        client_id=client_id, protocol=mqtt.MQTTv5)
        if username:
            c.username_pw_set(username, password)
        if use_tls:
            c.tls_set()
        # Backoff exponencial; um piso aleatório espalha frotas reconectando
        c.reconnect_delay_set(min_delay=random.randint(1, 5), max_delay=60)
        c.on_connect = self._on_connect
        c.on_disconnect = self._on_disconnect
        c.on_connect_fail = self._on_connect_fail
        c.on_message = self._on_message
        self.client = c

    def start(self):
        props = Properties(PacketTypes.CONNECT)
        props.SessionExpiryInterval = self.session_expiry_s
        self.client.connect_async(self.host, self.port, keepalive=30,
                                  clean_start=False, properties=props)
        self.client.loop_start()

    def stop(self):
        self.client.disconnect()
        self.client.loop_stop()

    # --- callbacks -------------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.warning("connect refused: %s", reason_code)
            return
        log.info("connected, session_present=%s", flags.session_present)
        if self.subscriptions:
            client.subscribe(self.subscriptions)  # sempre reassina
        self._flush()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            log.info("clean disconnect")
        else:
            log.warning("connection lost: %s, reconnecting", reason_code)

    def _on_connect_fail(self, client, userdata):
        log.warning("connection attempt failed, retrying with backoff")

    def _on_message(self, client, userdata, msg):
        if self.on_payload:
            self.on_payload(msg.topic, msg.payload)

    # --- publicação ------------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if self.client.is_connected():
            info = self.client.publish(topic, payload, qos=qos, retain=retain)
            if info.rc == mqtt.MQTT_ERR_SUCCESS:
                return True
        with self.lock:
            self.offline.append((topic, payload, qos, retain))
        return False

    def _flush(self):
        with self.lock:
            pending = list(self.offline)
            self.offline.clear()
        if pending:
            log.info("flushing %d buffered messages", len(pending))
        for topic, payload, qos, retain in pending:
            self.client.publish(topic, payload, qos=qos, retain=retain)


if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:8]
    prefix = f"connectmq/demo/{run_id}"
    rc = ResilientClient(
        host="broker.connectmq.com", port=1883,
        client_id=f"resilient-{run_id}",
        subscriptions=[(f"{prefix}/cmd/#", 1)],
    )
    rc.on_payload = lambda t, p: log.info("command on %s: %s", t, p)
    rc.start()
    print(f"publishing to {prefix}/telemetry, send commands to {prefix}/cmd/...")
    try:
        n = 0
        while True:
            n += 1
            payload = json.dumps({"seq": n, "ts": int(time.time() * 1000)})
            sent = rc.publish(f"{prefix}/telemetry", payload, qos=1)
            log.info("seq %d %s (buffered: %d)", n, "sent" if sent else "buffered", len(rc.offline))
            time.sleep(2)
    except KeyboardInterrupt:
        rc.stop()

Por que cada peça está ali

O número de sequência em cada payload facilita os testes: depois de uma queda, lacunas ou duplicatas em seq mostram exatamente o que aconteceu. Como o QoS 1 é entrega "pelo menos uma vez", os consumidores devem tolerar uma duplicata ocasional, por exemplo ignorando números de sequência que já viram. Nosso artigo sobre níveis de QoS do MQTT explica por que duplicatas são possíveis.

Como testar a lógica de reconexão

Não confie em código de reconexão que você não quebrou de propósito. Rode o script e um subscriber lado a lado:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-<run-id> -t "connectmq/demo/<run-id>/#" -v

Depois, teste cada modo de falha, um de cada vez:

1. Queda de rede. Desative o Wi-Fi ou desconecte o cabo por 30 segundos. O log mostra "connection lost", depois tentativas falhando com intervalos crescentes, depois "connected" e "flushing N buffered messages". O subscriber vê os números de sequência do buffer chegarem em uma rajada.
2. Porta bloqueada no Linux. Simule um broker inacessível enquanto o resto da rede funciona:

sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# espere 60 segundos e depois remova a regra
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Pacotes descartados são um teste mais duro do que uma conexão recusada, porque o cliente só percebe pelo timeout de keep-alive, cerca de uma vez e meia o intervalo de keep-alive. 3. Iniciar offline. Desconecte a rede antes de iniciar o script. Com connect_async(), ele espera e conecta quando a rede volta, em vez de sair. 4. Tomada de sessão. Inicie um segundo cliente com o mesmo client ID, por exemplo mosquitto_sub -i resilient-<run-id> -t x. O broker desconecta o cliente original, que reconecta e, por sua vez, desconecta o intruso. Esse é o sintoma dos "dois processos brigando"; vê-lo de propósito facilita reconhecê-lo depois. 5. Comandos durante uma queda. Enquanto o script estiver offline, publique um comando QoS 1 no topic cmd dele. Como a sessão foi mantida, o comando é entregue quando o cliente volta.

Reason codes que valem a pena tratar

O MQTT 5 informa por que uma conexão terminou. Registre o reason code em log e reaja aos que uma nova tentativa não vai resolver.

Tentar sem parar após uma recusa 0x86 só inunda os logs do broker, e muitos brokers, incluindo o ConnectMQ, bloqueiam um endereço IP após repetidas falhas de login. No _on_connect, considere chamar client.disconnect() e alertar um humano quando o código indicar um problema permanente.

Configurações para produção

Em implantações reais, conecte via TLS na porta 8883 com credenciais emitidas pelo seu broker. Com um broker MQTT gerenciado do ConnectMQ, construa o cliente assim:

rc = ResilientClient(
    host="broker.connectmq.com", port=8883, use_tls=True,
    client_id="YOUR_CLIENT_ID", username="YOUR_USERNAME", password="YOUR_PASSWORD",
    subscriptions=[("connectmq/factory1/production/line1/+/cmd", 1)],
)

Tenha alguns limites em mente. No servidor MQTT público, as sessões expiram em no máximo 30 minutos, então o expiry de 1.800 segundos usado acima é o valor máximo útil ali. O plano Serverless gratuito suporta QoS até 1, que é o que este cliente usa. Os planos pagos adicionam QoS 2 se você precisar de entrega exatamente uma vez. O broker também publica o estado das conexões em $SYS/broker/connection/{clientId}/state, o que ajuda a confirmar que um dispositivo voltou depois de uma queda.

Para um contexto mais amplo, nosso tutorial de Paho MQTT em Python cobre o básico, o guia da callback API do paho-mqtt 2.0 explica as assinaturas de callback da versão 2 usadas aqui, e a comparação de bibliotecas de cliente MQTT ajuda se você estiver escolhendo entre o paho e bibliotecas assíncronas. A documentação oficial do paho-mqtt lista todas as opções relacionadas à reconexão.

Comece a usar o ConnectMQ

Teste a sua lógica de reconexão hoje mesmo no servidor MQTT público gratuito e depois passe para credenciais privadas na implantação. O ConnectMQ oferece um broker MQTT gerenciado com MQTT 5, TLS, session expiry e reason codes claros, para que os seus clientes recebam um retorno útil quando algo der errado. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito.

Perguntas frequentes

O paho-mqtt reconecta automaticamente?

Sim, depois que uma conexão foi estabelecida e o network loop está rodando com loop_start() ou loop_forever(). Ele não tenta de novo uma primeira chamada connect() que falhou e não reassina por você.

Por que o meu cliente paho para de receber mensagens depois de reconectar?

As assinaturas se perderam junto com a sessão antiga. Mova as chamadas de subscribe() para o on_connect, para que rodem depois de cada conexão bem-sucedida.

Qual intervalo de reconexão devo usar?

Comece entre 1 e 5 segundos e limite entre 60 e 120 segundos. Aleatorizar o intervalo inicial por dispositivo evita que uma frota inteira reconecte no mesmo instante.

Devo usar clean_start=False?

Use, junto com um session expiry interval, quando quiser que o broker mantenha as assinaturas e enfileire mensagens QoS 1 durante quedas curtas. Mantenha o client ID estável, porque a sessão está vinculada a ele.