Teste de fan-in MQTT: milhares de dispositivos, um backend

Fan-in é o formato do dia a dia em IoT: milhares de dispositivos publicam cada um um pouco de telemetria, e um backend se inscreve em devices/+/telemetry para armazenar ou processar tudo. Individualmente os dispositivos são discretos; juntos, mandam tudo por uma inscrição, uma conexão TCP e um processo. Um teste de fan-in descobre quantos dispositivos esse consumidor consegue absorver antes de se formar um backlog — e como escalá-lo quando não consegue.

Este guia simula uma frota com emqtt-bench, mede backlog e atraso no lado do consumidor com um script Python e mostra como as shared subscriptions distribuem a carga. O padrão inverso é tratado no teste de fan-out.

Onde os gargalos do fan-in se escondem

O broker costuma lidar bem com a entrada do fan-in — rotear muitas mensagens pequenas para um subscriber é barato. O problema está mais adiante:

Os dois últimos são limites do broker sobre uma única sessão — por exemplo max_queued_messages e max_inflight_messages no Mosquitto. Quando o consumidor fica para trás, o broker enfileira mensagens para ele e, quando a fila enche, começa a descartar ou desconectar, dependendo do broker. Veja consumidores lentos para saber como cada broker se comporta.

Passo 1: simular a frota

O emqtt-bench consegue fazer o papel de milhares de dispositivos a partir de uma máquina. Dê a cada cliente o próprio topic com %i (o índice do cliente) para que a estrutura de topics corresponda à produção. O intervalo -I é por cliente e em milissegundos, então 5.000 dispositivos com -I 5000 enviam uma mensagem a cada 5 segundos cada — 1.000 msg/s no total:

docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 5000 -i 5 -I 5000 \
  -t 'testmqtt/fanin/%i/telemetry' -s 200 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

Para aumentar a carga, reduza -I (5000 → 2000 → 1000 ms dá 2.500 e 5.000 msg/s) em vez de adicionar conexões: você está testando o consumidor, não o número de conexões. Antes de começar, confirme com o topic matcher que o seu wildcard realmente casa com os topics dos dispositivos — um erro no filtro parece exatamente um consumidor que não recebe nada.

Passo 2: um consumidor que mostra o backlog

Um consumidor de produção nunca deve fazer trabalho lento dentro do on_message: isso bloqueia o loop de rede, que para de ler o socket e empurra o backlog para o broker, onde você não o vê. Este consumidor com paho-mqtt 2.x entrega cada mensagem a uma fila, processa em threads de trabalho com um custo simulado por mensagem e imprime a cada segundo a taxa de recebimento, a taxa de processamento e o backlog:

import json, queue, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = "testmqtt/fanin/+/telemetry"
SHARED = "--shared" in sys.argv
TOPIC = f"$share/ingest/{FILTER}" if SHARED else FILTER
WORK_MS = 0.5        # custo simulado por mensagem (parse + insert no banco)
WORKERS = 4

inbox = queue.Queue(maxsize=100_000)
counts = {"rx": 0, "done": 0}
lags = []

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

def on_message(client, userdata, msg):
    counts["rx"] += 1
    inbox.put(msg.payload)       # nunca faça trabalho lento na thread de rede

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # mensagens de sonda com timestamp
            try:
                lags.append((time.time() - json.loads(payload)["t"]) * 1000)
            except (ValueError, KeyError):
                pass
        time.sleep(WORK_MS / 1000)
        counts["done"] += 1

for _ in range(WORKERS):
    threading.Thread(target=worker, daemon=True).start()

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                client_id=f"ingest-{uuid.uuid4().hex[:6]}", protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect(HOST, PORT)
c.loop_start()

while True:
    time.sleep(1)
    l = sorted(lags); lags.clear()
    p99 = f"{l[int(len(l) * 0.99) - 1]:.0f} ms" if l else "-"
    print(f"rx {counts['rx']:>6}/s  done {counts['done']:>6}/s  "
          f"backlog {inbox.qsize():>6}  probe lag p99 {p99}")
    counts.update(rx=0, done=0)

Os payloads do emqtt-bench são bytes aleatórios, então não carregam timestamp. Para medir o atraso ponta a ponta, rode uma pequena sonda na mesma máquina do consumidor (para os dois usarem o mesmo relógio) que publique JSON com timestamp na mesma árvore de topics:

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

p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
p.connect("broker.connectmq.com", 1883)
p.loop_start()
while True:
    p.publish("testmqtt/fanin/probe/telemetry", json.dumps({"t": time.time()}), qos=1)
    time.sleep(0.1)

Lendo os números

Rode a frota em cada degrau e acompanhe a saída do consumidor por alguns minutos:

• rx ≈ done, backlog perto de zero, atraso estável — o consumidor está dando conta. Vá para o próximo degrau.
• rx ≈ entrada mas done < rx, backlog crescendo — a parte de rede está bem, o gargalo é o processamento. Adicione workers, agrupe escritas ou acelere o destino.
• rx < entrada, backlog pequeno — o limite é a conexão única ou a sessão no broker (janela inflight, fila por cliente). É aqui que uma shared subscription ajuda.
• Atraso subindo sem parar — você está acima da taxa sustentável mesmo que nada se perca ainda; a fila só está ficando maior.
• Consumidor desconectado sob carga — o broker atingiu o limite por cliente ou o keep-alive expirou porque a thread de rede estava bloqueada.

Passo 3: escalar com shared subscriptions

Quando o limite é uma única conexão de consumidor, inicie vários consumidores no mesmo grupo de shared subscription. O broker então entrega cada mensagem a apenas um membro de $share/ingest/...:

python ingest.py --shared   # terminal 1
python ingest.py --shared   # terminal 2 (e assim por diante)

Repita o degrau que falhou antes. O que verificar:

1. A soma de done entre os consumidores bate com a entrada da frota — nenhuma mensagem perdida.
2. Sem duplicatas — sem $share, cada consumidor receberia todas as mensagens e você processaria tudo duas vezes. Conte IDs únicos se o payload os tiver.
3. Distribuição equilibrada — os brokers usam estratégias diferentes (round robin, aleatória, sticky por cliente). Confira se a carga está razoavelmente equilibrada.
4. Failover — mate um consumidor durante a execução; os outros devem absorver a parte dele. Mensagens em voo para o consumidor derrubado podem ser reenviadas ou perdidas, dependendo do QoS e do broker.
5. Ordem — a ordem por dispositivo não é garantida ao longo do grupo. Se importar, projete para isso.

Shared subscriptions explica a semântica, e o teste de shared subscriptions cobre em mais detalhes os testes de distribuição e failover.

Além de um processo

Se o gargalo é o destino e as rajadas são inevitáveis — uma frota que reconecta depois de uma queda reenvia de uma vez muitos dados armazenados —, coloque um buffer durável entre o MQTT e o armazenamento. Fazer a ponte de MQTT para Kafka é um padrão comum: veja MQTT para Kafka. Depois teste explicitamente o caso de rajada: pare o consumidor por cinco minutos enquanto a frota publica, inicie-o de novo e meça quanto tempo o backlog leva para esvaziar. Para um comportamento de dispositivo mais realista do que o do emqtt-bench, use um simulador de dispositivos com os seus payloads reais.

Antes de uma execução grande, uma checagem rápida no cliente MQTT online — inscreva-se em testmqtt/fanin/# e veja as mensagens chegando — confirma que a frota está publicando onde você imagina.

O fan-in raramente quebra o broker; ele quebra o consumidor. Meça backlog e atraso em vez de taxas brutas de recebimento, tire o trabalho lento da thread de rede e escale com shared subscriptions quando uma conexão não for suficiente.

Perguntas frequentes