Como testar assinaturas compartilhadas e balanceamento de carga no MQTT

Uma assinatura compartilhada ($share/<grupo>/<topic>) transforma vários subscribers num pool de workers: cada mensagem do topic é entregue a um único membro do grupo em vez de a todos. Para testar, suba vários workers no mesmo grupo, publique uma sequência numerada e verifique três coisas: cada mensagem chegou exatamente uma vez ao grupo, a carga foi distribuída e nada se perdeu quando um worker caiu no meio do fluxo.

Este guia mostra uma checagem rápida pela CLI, um teste em Python com simulação de queda e os casos de borda que vale a pena validar. Os conceitos estão em assinaturas compartilhadas no MQTT. Aqui a gente só testa.

O que um broker correto faz

A estratégia de distribuição fica a critério do broker. O Mosquitto usa round robin. Outros brokers oferecem estratégias aleatória, sticky ou baseada em hash, muitas vezes configuráveis. Seu teste deve verificar uma distribuição justa dentro de uma tolerância, não um padrão exato, a menos que você dependa de uma estratégia específica.

Passo 0: verificar se o broker suporta

Um broker MQTT 5 anuncia o suporte no CONNACK com a propriedade Shared Subscription Available. Se a propriedade não vier, o suporte é presumido. Se vier 0, o broker vai rejeitar ou ignorar assinaturas $share, e o SUBACK retorna 0x9E Shared Subscriptions not supported. De qualquer forma, a prova real é o teste de distribuição mais abaixo. A consulta de reason codes decifra o SUBACK que você receber.

O nome do grupo é um único nível de topic: sem /, + ou #. O resto é um filtro de topic normal, então wildcards funcionam depois do grupo: $share/workers/testmqtt/share/+/jobs. Confira os filtros com o topic matcher antes de culpar o broker.

Checagem rápida com a CLI do Mosquitto

Abra dois terminais com membros do mesmo grupo e um terceiro com um subscriber normal:

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t '$share/demo/testmqtt/share/9b1e/jobs' -v

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
    -t testmqtt/share/9b1e/jobs -m "job $i"
done

Os terminais 1 e 2 dividem os dez jobs entre si (com round robin, ímpares de um lado e pares do outro). O terminal 3 vê todos os dez. Sempre coloque o topic entre aspas simples: sem aspas ou com aspas duplas, $share seria expandido pelo shell como variável.

Teste automatizado de distribuição e failover

Este script com paho-mqtt 2.x sobe três workers num grupo, publica 100 mensagens QoS 1 numeradas e informa a contagem por worker, lacunas e duplicatas. Com --failover ele derruba o worker-0 na metade: o worker para de ler, então suas últimas mensagens ficam sem confirmação, e em seguida a conexão TCP é cortada sem DISCONNECT, como numa queda real.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

Uma execução contra um broker com round robin fica assim:

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

Como interpretar:

• Sem --failover, as contagens devem ser parecidas (33/33/34 com round robin). Uma divisão muito desigual indica estratégia sticky ou por hash, ou um worker lento que o broker está pulando.
• Com --failover, missing 0 significa que o broker redirecionou o trabalho do membro que caiu. IDs faltando bem em volta do momento da queda significam que mensagens em trânsito foram perdidas junto com a sessão. Alguns brokers esperam primeiro o cliente reconectar, então teste com suas configurações de sessão reais.
• Duplicatas são legais no QoS 1: uma mensagem pode ser entregue ao worker que caiu e reenviada a outro. Workers num grupo compartilhado precisam ser idempotentes. Veja mensagens MQTT duplicadas.
• Mensagens faltando sem nenhuma queda geralmente indicam limite de taxa ou de fila no broker. Diminua RATE e rode de novo. Endpoints públicos compartilhados são especialmente propensos a limitar.

Casos de borda que vale validar

1. Ordem. Cada membro vê as próprias mensagens em ordem, mas não há ordem global no grupo. A mensagem 41 pode ser processada antes da 40 em outro worker. Se a ordem importa por dispositivo, use uma estratégia por hash de topic quando existir ou particione os topics. O teste de ordem de mensagens mostra como medir isso.
2. Mensagens retidas. Publique uma mensagem retida e depois suba um subscriber compartilhado: ele não deve receber nada. Workers que precisam de estado inicial devem obtê-lo de outra forma.
3. QoS misto. O QoS entregue é o mínimo entre o QoS de publicação e o QoS concedido a cada membro. Um membro assinado em QoS 0 perde mensagens ao cair, mesmo que os outros usem QoS 1.
4. Opções de assinatura. No Local numa assinatura compartilhada é erro de protocolo. As demais opções estão explicadas em opções de assinatura MQTT.
5. Escalar o grupo. Adicione um quarto worker no meio da execução: ele deve começar a receber sua parte na hora. Remova um: os outros absorvem a carga.
6. Membros persistentes. Com Session Expiry maior que 0, um membro desconectado mantém o lugar no grupo, e alguns brokers continuam enviando mensagens que se acumulam enquanto ele está offline. Decida se os workers devem mesmo ser persistentes.

Testar o ganho de throughput

Assinaturas compartilhadas existem para escalar consumidores. Meça se o ganho é real: rode o script com 1, 2 e 4 workers que esperam um tempo fixo de processamento por mensagem e compare o tempo total para esvaziar uma rajada. O throughput deve crescer quase linearmente até o broker ou a rede virarem o limite. Um teste de fan-in é o complemento natural: muitos publishers alimentando um grupo.

Ver a distribuição ao vivo

Abra o cliente MQTT online em duas abas do navegador, assine as duas no mesmo filtro $share/... e publique de uma terceira aba. Você vê a divisão em tempo real, uma boa checagem antes de rodar o script.

Publicar jobs de teste pelo navegador

Um teste de assinaturas compartilhadas leva poucos minutos e responde às perguntas que importam em produção: todo job é processado, a carga é realmente distribuída e o que acontece com o trabalho em andamento quando um worker morre. Mantenha a execução de failover na sua suíte de CI e repita sempre que trocar de broker ou de configuração de sessão.

Perguntas frequentes