Para testar as garantias de entrega do MQTT, publique mensagens com número de sequência, derrube a conexão enquanto há mensagens em trânsito e conte o que chega. O QoS 1 passa quando nenhum número de sequência está faltando (duplicatas são aceitáveis). O QoS 2 passa quando nada falta e nada se repete. Todo o resto — sessões persistentes, pacotes de confirmação, limites de fila do broker — serve para tornar esse experimento honesto.
Este guia mostra como observar o fluxo de confirmações, um harness em Python que conta lacunas e duplicatas, as falhas que vale a pena injetar e como interpretar os resultados.
O que cada nível promete
| QoS 0 | QoS 1 | QoS 2 | |
|---|---|---|---|
| Garantia | No máximo uma vez | Pelo menos uma vez | Exatamente uma vez |
| Pacotes por salto | PUBLISH | PUBLISH → PUBACK | PUBLISH → PUBREC → PUBREL → PUBCOMP |
| Depois de uma queda de conexão | A mensagem pode se perder | Reenviada com DUP=1, duplicatas possíveis | Retomada a partir do último passo, sem duplicatas |
| O teste passa quando | — (perda é permitida) | lacunas = 0 | lacunas = 0 e duplicatas = 0 |
Dois detalhes decidem a maioria dos resultados. Primeiro, o QoS é por salto: o QoS do publisher cobre publisher → broker, o QoS da assinatura cobre broker → assinante, e o QoS efetivo de entrega é o menor dos dois. Um publish em QoS 2 para uma assinatura QoS 1 é uma entrega QoS 1. Segundo, as garantias só sobrevivem a uma desconexão se a sessão sobreviver — caso contrário o broker joga fora o estado em trânsito. O guia de níveis de QoS cobre a teoria; aqui nós a testamos.
Passo 1: observe as confirmações
Antes de automatizar qualquer coisa, olhe os pacotes. Os clientes do Mosquitto imprimem cada pacote de controle com -d:
# Terminal 1: subscribe at QoS 2 with debug output
mosquitto_sub -h broker.connectmq.com -p 1883 -i testmqtt-qos-sub -t 'testmqtt/qos/demo' -q 2 -d
# Terminal 2: publish one QoS 2 message with debug output
mosquitto_pub -h broker.connectmq.com -p 1883 -i testmqtt-qos-pub -t 'testmqtt/qos/demo' -m 'hello' -q 2 -dO publisher imprime um fluxo parecido com este:
Client testmqtt-qos-pub sending CONNECT
Client testmqtt-qos-pub received CONNACK (0)
Client testmqtt-qos-pub sending PUBLISH (d0, q2, r0, m1, 'testmqtt/qos/demo', ... (5 bytes))
Client testmqtt-qos-pub received PUBREC (Mid: 1)
Client testmqtt-qos-pub sending PUBREL (m1)
Client testmqtt-qos-pub received PUBCOMP (Mid: 1)
Client testmqtt-qos-pub sending DISCONNECTd0 é a flag DUP, q2 o QoS e m1 o packet identifier. Repita com -q 1 e você deve ver um único PUBACK no lugar do handshake de três etapas. No lado do assinante, confira o valor q no PUBLISH recebido — se você assinou em QoS 1, ele mostra q1 mesmo que o publisher tenha usado QoS 2.
Para uma inspeção mais profunda, capture o tráfego e filtre no Wireshark com mqtt.msgtype >= 3 && mqtt.msgtype <= 7 (de PUBLISH a PUBCOMP) ou mqtt.dupflag == 1 para ver só as retransmissões. No MQTT 5, as confirmações também trazem um reason code: um PUBACK com 0x10 (No matching subscribers) é sucesso, não erro — consulte os códigos na referência de reason codes.
Passo 2: um contador de lacunas e duplicatas
O harness tem três partes: um assinante com sessão persistente que registra cada número de sequência recebido, um publisher que envia um intervalo conhecido e conta as confirmações, e um pequeno script de análise.
import sys
import paho.mqtt.client as mqtt
QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
TOPIC = "testmqtt/qos-test/seq"
log = open(f"received-q{QOS}.log", "a", buffering=1) # line-buffered: survives kill -9
def on_connect(client, userdata, flags, reason_code, properties):
print(f"connected: {reason_code}, session present: {flags.session_present}")
client.subscribe(TOPIC, qos=QOS)
def on_message(client, userdata, msg):
log.write(f"{msg.payload.decode()} {int(msg.dup)}\n")
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"testmqtt-qos-sub-q{QOS}", clean_session=False)
# client.username_pw_set("username", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_forever() # reconnects automatically after a network errorimport sys, time
import paho.mqtt.client as mqtt
QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
N = int(sys.argv[2]) if len(sys.argv) > 2 else 2000
TOPIC = "testmqtt/qos-test/seq"
acked = 0
def on_publish(client, userdata, mid, reason_code, properties):
global acked
acked += 1 # QoS 1: on PUBACK, QoS 2: on PUBCOMP
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"testmqtt-qos-pub-q{QOS}", clean_session=False)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_start()
for seq in range(N):
client.publish(TOPIC, str(seq), qos=QOS) # queued by paho while disconnected
time.sleep(0.005)
deadline = time.time() + 60
while acked < N and time.time() < deadline:
time.sleep(0.2)
print(f"published {N}, acknowledged {acked}")
client.disconnect()
client.loop_stop()#!/usr/bin/env bash
LOG=${1:-received-q1.log}; N=${2:-2000}
echo "received lines : $(wc -l < "$LOG")"
echo "unique : $(cut -d' ' -f1 "$LOG" | sort -u | wc -l)"
echo "duplicates : $(cut -d' ' -f1 "$LOG" | sort | uniq -d | wc -l)"
echo "with DUP flag : $(awk '$2 == 1' "$LOG" | wc -l)"
echo "missing : $(comm -23 <(seq 0 $((N - 1)) | sort) <(cut -d' ' -f1 "$LOG" | sort -u) | wc -l)"Rode python sub.py 1, depois python pub.py 1 e então ./analyze.sh received-q1.log. Comece com uma execução sem falhas: ela precisa mostrar 2.000 únicos e 0 faltando. Se a linha de base já falha, corrija isso primeiro (QoS errado na assinatura, uma ACL, um limite de tamanho de mensagem) antes de injetar falhas.
Passo 3: quebre coisas no meio do envio
Rode cada cenário enquanto o pub.py está enviando e depois analise. Use um arquivo de log novo a cada execução.
| Cenário | Como | QoS 1 deve mostrar | QoS 2 deve mostrar |
|---|---|---|---|
| Linha de base | Sem falha | 0 faltando | 0 faltando, 0 duplicatas |
| Resetar as conexões | sudo ss -K dst <broker-ip> dport = 1883 (Linux) | 0 faltando, algumas duplicatas | 0 faltando, 0 duplicatas |
| Buraco negro por 20 s | Regra DROP do iptables e depois removê-la | 0 faltando, duplicatas prováveis | 0 faltando, 0 duplicatas |
| Latência e resets | Toxics latency e reset_peer do Toxiproxy | 0 faltando | 0 faltando, 0 duplicatas |
| Queda do assinante | kill -9 no sub.py e reiniciá-lo em poucos segundos | 0 faltando, duplicatas prováveis | Veja abaixo |
BROKER_IP=$(dig +short broker.connectmq.com | head -1)
sudo iptables -I OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP
sleep 20
sudo iptables -D OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROPO buraco negro é a falha mais realista: nenhum RST chega, então o cliente só percebe pelo keep-alive (cerca de 1,5 × os 10 segundos de keep-alive no lado do broker) e então reconecta. Na reconexão o assinante deve imprimir session present: True, e o broker deve reenviar as mensagens não confirmadas com DUP=1, seguidas da fila acumulada. Mais técnicas de corte em testar QoS com corte de rede.
A queda do processo é o caso interessante do QoS 2. O paho-mqtt guarda o estado de sessão em memória, então um kill -9 apaga a metade do handshake QoS 2 que pertence ao cliente. Dependendo de a biblioteca entregar a mensagem ao seu código no PUBLISH ou no PUBREL, uma queda no momento errado pode gerar uma duplicata ou uma mensagem perdida. Exactly-once é uma propriedade do protocolo entre duas pontas que mantêm seu estado — se a sua aplicação não pode processar duas vezes após uma queda, deduplique por um ID de mensagem no seu próprio armazenamento.
Interpretando os resultados
- Mensagens faltando em QoS 1/2 após um corte de rede — a sessão não persistiu: confira
session presentno log, o client ID (precisa ser idêntico na reconexão) e o session expiry do MQTT 5. - Mensagens faltando só após quedas longas — a fila por cliente do broker encheu e ele descartou mensagens por política (por exemplo,
max_queued_messagesno Mosquitto). É um limite de capacidade, não um bug de QoS; veja fila de mensagens cheia. - Duplicatas em QoS 2 — confira o QoS efetivo no assinante (
q2nomosquitto_sub -d) e procure quedas ou um segundo cliente usando o mesmo client ID. Mais causas em mensagens duplicadas. - Duplicatas sem flag DUP em QoS 1 — normalmente o publisher reenviou após reconectar, o que para o broker é uma mensagem nova. Normal no at-least-once.
acknowledgedmenor quepublished— o publisher desistiu antes de a fila esvaziar; aumente o prazo ou verifique se há um limite de inflight no broker.
Uma verificação manual rápida
Para uma checagem rápida sem scripts, assine em QoS 1 no cliente MQTT online, publique de um terminal com mosquitto_pub -q 1 -d e compare o QoS exibido nos dois lados. Depois desconecte o cliente do navegador, publique mais algumas mensagens e reconecte com o mesmo client ID e clean session desativado: as mensagens enfileiradas devem chegar.
Veja entregas QoS no seu navegador
O QoS 1 se prova com zero lacunas e o QoS 2 com zero lacunas e zero duplicatas — medidos com números de sequência através de desconexões reais, não presumidos a partir da especificação. Mantenha o harness no seu repositório e rode-o de novo sempre que mudar o broker, a biblioteca cliente ou as configurações de sessão.
Perguntas frequentes
Como provar que o MQTT QoS 1 não perde mensagens?
Publique mensagens numeradas em QoS 1 para um assinante com sessão persistente, derrube a conexão no meio do teste e compare os números de sequência recebidos com o intervalo enviado. Zero lacunas prova o at-least-once; duplicatas são permitidas e esperadas.
Por que recebo duplicatas com QoS 2?
Geralmente porque um dos saltos não é realmente QoS 2 (a assinatura foi feita em QoS 1, então a entrega é rebaixada), porque a sessão não era persistente ou porque o processo cliente caiu e perdeu o estado de sessão em memória. Exactly-once só vale entre pontas que preservam seu estado de sessão.
Preciso de uma sessão persistente para testar QoS?
Sim, em qualquer teste que envolva uma desconexão. Sem sessão persistente (clean session, ou MQTT 5 com session expiry 0) o broker descarta as mensagens em trânsito e na fila quando a conexão cai, e você acaba testando o tratamento de sessão, não o QoS.
