Garantias de Ordem de Mensagens no MQTT Explicadas

A ordem de mensagens MQTT é garantida, mas apenas dentro de limites específicos que muitos desenvolvedores nunca leem. Mensagens de um publicador em um topic, com o mesmo nível de QoS, chegam em ordem; quase tudo além disso não é prometido. Este artigo explica exatamente o que a especificação garante, as situações comuns em que as mensagens parecem fora de ordem e como adicionar números de sequência para que seus consumidores consigam detectar e tratar isso.

O que a especificação MQTT garante

Tanto o MQTT 3.1.1 quanto o MQTT 5 definem um "Ordered Topic". Por padrão, o broker deve tratar todo topic como ordenado, o que leva a esta garantia:

Mensagens publicadas por um client em um topic, com o mesmo nível de QoS, são entregues a cada assinante na ordem em que foram publicadas.

A especificação também exige que, quando um client reenvia mensagens QoS 1 e QoS 2 não confirmadas após uma reconexão, ele o faça na ordem original, e que as confirmações sejam enviadas na ordem em que os pacotes foram recebidos.

Leia essa garantia com atenção e as lacunas ficam óbvias. Ela não diz nada sobre:

• Mensagens de publicadores diferentes no mesmo topic.
• Mensagens de um publicador em topics diferentes.
• Mensagens de um publicador no mesmo topic, mas com níveis de QoS diferentes.
• O que a sua aplicação faz depois que a biblioteca do client entrega a mensagem.

Onde a ordem de mensagens MQTT deixa de valer

Duas dessas merecem um olhar mais atento.

Shared subscriptions. Com $share/group/topic, o broker distribui as mensagens entre vários consumidores para balanceamento de carga. Cada consumidor vê um subconjunto, e eles processam em paralelo, então não há uma ordem global significativa. Se a ordem por dispositivo importa, projete para que as mensagens de cada dispositivo sempre cheguem ao mesmo consumidor, ou reordene mais adiante. Nosso guia de shared subscriptions MQTT aborda essa troca.

Reentrega e duplicatas. QoS 1 é at-least-once. Se uma confirmação se perde, o remetente retransmite com a flag DUP ativada. Um assinante pode então ver a mensagem 5 de novo depois de já ter processado a 6 e a 7. Isso é uma duplicata, e não uma reordenação de verdade, mas para um código ingênuo parece que o tempo voltou. A solução é abordada em como tratar mensagens duplicadas no MQTT.

Demonstrando a ordem no servidor MQTT público

O experimento mais rápido usa o servidor MQTT público (gratuito) em broker.connectmq.com, que serve apenas para testes e protótipos. Observe o limite de 10 mensagens por segundo por client; por isso o publicador abaixo se autolimita.

Assinante:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ord-sub-h5d3 -q 1 \
  -t 'connectmq/demo/h5d3/#' -v

Publicador, um topic, QoS 1, mensagens numeradas:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i ord-pub-h5d3 -q 1 \
    -t connectmq/demo/h5d3/seq -m "$i"
  sleep 0.2
done

Você verá de 1 a 20 em ordem. Note que cada chamada de mosquitto_pub é uma conexão separada, então é o loop sequencial que mantém a ordem aqui. Agora rode dois loops em paralelo em dois topics diferentes, seq-a e seq-b. Cada topic continua em ordem, mas as linhas dos dois topics se intercalam de forma imprevisível. Não há nada de errado: a ordem entre topics simplesmente não faz parte do contrato.

Adicionando números de sequência

Quando a ordem importa para a sua aplicação, transporte-a no payload. Um número de sequência monotônico por publicador, mais um timestamp, permite que os consumidores detectem lacunas, duplicatas e mensagens obsoletas. Aqui está um par completo de publicador e consumidor com paho-mqtt 2.x:

import json
import time
import uuid
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
TOPIC = "connectmq/demo/h5d3/pump-07/telemetry"
last_seq = {}  # dispositivo -> maior sequência vista

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    dev, seq = data["device"], data["seq"]
    prev = last_seq.get(dev)
    if prev is not None and seq <= prev:
        print(f"stale or duplicate seq={seq} (last={prev}), dropped")
        return
    if prev is not None and seq > prev + 1:
        print(f"gap: missing {prev + 1}..{seq - 1}")
    last_seq[dev] = seq
    print(f"apply seq={seq} value={data['value']}")

sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"ord-c-{uuid.uuid4().hex[:6]}")
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(BROKER, 1883, keepalive=30)
sub.loop_start()
time.sleep(1)

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"ord-p-{uuid.uuid4().hex[:6]}")
pub.connect(BROKER, 1883, keepalive=30)
pub.loop_start()
for seq in [1, 2, 3, 5, 4, 6]:  # o 4 é enviado atrasado de propósito
    body = {"device": "pump-07", "seq": seq, "ts": time.time(), "value": 20 + seq}
    pub.publish(TOPIC, json.dumps(body), qos=1).wait_for_publish()
    time.sleep(0.2)

time.sleep(2)
for c in (pub, sub):
    c.loop_stop()
    c.disconnect()

O consumidor aplica 1, 2, 3, informa uma lacuna antes do 5, descarta o 4 atrasado como obsoleto e depois aplica o 6. Se mensagens atrasadas devem ser descartadas, armazenadas em buffer ou aceitas depende dos dados, o que nos leva à estratégia.

Escolhendo uma estratégia de ordenação

1. O último valor vence (estado). Para leituras como a temperatura de uma câmara fria em um frigorífico, descarte qualquer coisa mais antiga do que o que você já tem. O código acima faz isso.
2. Buffer de reordenação (fluxos de eventos). Segure as mensagens por um breve período e libere-as na ordem de sequência, desistindo das lacunas após um timeout. Útil quando cada evento importa, como em contadores de produção de uma linha de envase.
3. Upserts idempotentes (armazenamento). Grave cada mensagem com chave por dispositivo e sequência ou timestamp. O banco de dados termina correto independentemente da ordem de chegada.
4. Um topic por fluxo ordenado. Se os comandos precisam ser aplicados em ordem, publique-os em um único topic, a partir de um único client, com um único QoS.
5. Evite misturar QoS em um mesmo fluxo. Escolha um QoS por topic.

Ordem em reconexões e períodos offline

As reconexões são onde as suposições sobre ordem mais são postas à prova. Quando um assinante com sessão persistente volta, o broker primeiro reenvia as mensagens que estavam em trânsito e não confirmadas, depois entrega as mensagens enfileiradas enquanto o client estava offline e então continua com o tráfego ao vivo. Dentro de um topic, vindo de um publicador, a ordem continua valendo, mas o consumidor de repente recebe uma rajada de mensagens antigas seguida das atuais.

Duas coisas ajudam aqui. Primeiro, mantenha o processamento por dispositivo sequencial, para que a rajada seja aplicada em ordem em vez de disputada por workers paralelos. Segundo, use timestamps no payload para decidir como apresentar os dados de recuperação: um dashboard pode querer pular direto para o valor mais recente, enquanto um historiador deve armazenar cada leitura com seu horário original. Se as mensagens enfileiradas só são úteis por um tempo limitado, um message expiry do MQTT 5 mantém as muito antigas fora da rajada.

Do lado do publicador, dispositivos que armazenam leituras localmente durante uma queda de rede devem enviar o buffer em ordem de sequência antes de retomar a publicação ao vivo, e no mesmo topic e QoS do tráfego normal.

Armadilhas de ordenação e solução de problemas

A sequência reinicia após um reboot. Um dispositivo que recomeça em seq 1 parece estar enviando dados obsoletos. Inclua um boot ID ou um contador de época junto com a sequência, ou persista o contador na flash.

A ordenação baseada em relógio falha. Os relógios dos dispositivos derivam e às vezes voltam para 1970. Prefira números de sequência para ordenar e timestamps para exibir.

A ordem se perde dentro da sua aplicação. Handlers assíncronos, pools de threads e filas no consumidor reordenam livremente. Processe o fluxo de cada dispositivo em um único worker ou particione por ID de dispositivo.

Efeitos da janela de mensagens em trânsito. O receive maximum do MQTT 5 permite que várias mensagens QoS 1 estejam em trânsito ao mesmo tempo. A ordem na rede continua valendo, mas as retentativas com perda de pacotes tornam duplicatas mais prováveis. Veja o artigo sobre o receive maximum do MQTT para ajustes.

Ordem em um broker MQTT gerenciado

O ConnectMQ segue as regras de ordem do MQTT por topic para clients MQTT 3.1.1 e 5.0. As shared subscriptions no plano Serverless (Paid) oferecem escalabilidade horizontal para os consumidores, com a mesma troca de ordenação descrita acima. Usar um broker MQTT gerenciado não muda as garantias do protocolo, mas elimina o trabalho operacional para que você se concentre no design dos consumidores. Para o contexto mais amplo do protocolo, veja o tutorial de MQTT.

Comece a usar o ConnectMQ

Crie um broker no plano Serverless (Free) em app.connectmq.com, sem cartão de crédito. Rode o exemplo de números de sequência com suas próprias credenciais do ConnectMQ, acompanhe o fluxo no explorador de topics ao vivo do portal e depois leve-o para produção em um broker MQTT na nuvem que escala junto com a sua frota.

Perguntas frequentes

O MQTT garante a ordem das mensagens?

Sim, para mensagens de um client em um topic com o mesmo QoS. Não há garantia entre topics, publicadores ou níveis de QoS misturados.

O QoS 2 garante a ordem?

O QoS 2 garante entrega exactly-once por salto. A ordem continua seguindo a mesma regra por topic e por publicador.

Por que vejo uma mensagem mais antiga depois de uma mais nova?

Geralmente é uma reentrega de QoS 1, um segundo publicador ou processamento paralelo no seu consumidor. Números de sequência facilitam identificar qual é o caso.

As shared subscriptions são ordenadas?

Não globalmente. Cada mensagem vai para um membro do grupo, e os membros processam em paralelo.