Teste de fan-out MQTT: um publisher, milhares de subscribers

Em um cenário de fan-out, uma mensagem vai para muitos receptores: uma mudança de configuração enviada para toda a frota, uma atualização de preço exibida em 10.000 dashboards, um aviso de "firmware disponível". O publisher envia um PUBLISH; o broker envia um para cada inscrição correspondente. Um teste de fan-out mede quanto tempo o último subscriber leva para receber a mensagem e se todos a receberam.

Este guia monta um benchmark 1 → 10.000, adiciona uma sonda que mede a dispersão de entrega e faz a conta de largura de banda para que a fatura de rede não surpreenda. Para testes gerais de taxa, veja o teste de throughput MQTT; para o padrão inverso, o teste de fan-in.

Por que fan-out é um teste à parte

Testes de throughput de publish costumam parear cada publisher com um subscriber. O fan-out estressa outra coisa: a capacidade do broker de copiar uma mensagem para milhares de filas de saída e despejá-las em milhares de sockets.

A pergunta não é "quantas mensagens por segundo", e sim "quanto tempo até todos os subscribers terem a mensagem" — a dispersão de entrega — no tamanho real da sua frota.

Primeiro, a conta de largura de banda

Faça a conta antes de começar: um teste de fan-out pode saturar um link de rede muito antes de o broker sofrer. Um PUBLISH na rede é aproximadamente cabeçalho fixo + 2 bytes de tamanho do topic + topic + identificador de pacote (QoS > 0) + propriedades + payload. Veja o overhead de banda do MQTT para os detalhes.

Exemplo: payload de 500 bytes em testmqtt/fanout/broadcast (25 bytes) com QoS 0 ≈ 530 bytes por PUBLISH.

Some ainda o overhead de TCP/IP e TLS. A última linha é mais do que muitas máquinas geradoras de carga conseguem receber sozinhas, e por isso testes de fan-out geralmente são distribuídos em várias máquinas. A calculadora de tamanho de pacote dá o tamanho exato por pacote para o seu topic e payload.

Passo 1: abrir os subscribers

O emqtt-bench consegue manter milhares de conexões de subscriber. Divida 10.000 entre quatro máquinas (ou containers em hosts diferentes) para ficar longe dos limites de file descriptors e de portas efêmeras:

ulimit -n 65535
docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench sub \
  -h broker.connectmq.com -p 1883 \
  -c 2500 -i 5 \
  -t testmqtt/fanout/broadcast -q 0 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

-i 5 abre uma conexão a cada 5 ms (200 por segundo por máquina), então as quatro máquinas precisam de pouco mais de 12 segundos para conectar todos. Abrir conexões mais rápido é um teste por si só — veja o teste de estresse de conexões — e não deve ser misturado com as medições de fan-out. Espere as quatro reportarem o total antes de publicar.

Passo 2: publicar e sondar

A saída de subscriber do emqtt-bench mostra a taxa agregada de recebimento, mas não o quanto o subscriber mais lento atrasou. Este script com paho-mqtt 2.x adiciona 50 subscribers sonda em um processo, publica mensagens com timestamp no mesmo topic de broadcast e reporta latência e dispersão de entrega entre as sondas. Como o publisher e as sondas compartilham o mesmo relógio, os números são diretamente comparáveis:

import json, threading, time, uuid
from collections import defaultdict
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = "testmqtt/fanout/broadcast"
PROBES, MESSAGES, RATE, QOS = 50, 60, 1, 0
lat = defaultdict(list)          # seq -> latências (ms) entre as sondas
lock = threading.Lock()

def on_message(client, userdata, msg):
    try:
        m = json.loads(msg.payload)
    except ValueError:
        return                   # ignorar tráfego que não é da sonda
    with lock:
        lat[m["seq"]].append((time.perf_counter() - m["t"]) * 1000)

def new_client(name):
    return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                       client_id=f"{name}-{uuid.uuid4().hex[:6]}")

probes = []
for n in range(PROBES):
    c = new_client(f"fanout-probe-{n}")
    c.on_message = on_message
    c.connect(HOST, PORT)
    c.subscribe(TOPIC, qos=QOS)
    c.loop_start()
    probes.append(c)

pub = new_client("fanout-pub")
pub.connect(HOST, PORT)
pub.loop_start()
time.sleep(3)                    # deixar as inscrições se estabilizarem

for seq in range(MESSAGES):
    body = {"seq": seq, "t": time.perf_counter(), "pad": "x" * 450}
    pub.publish(TOPIC, json.dumps(body), qos=QOS)
    time.sleep(1 / RATE)
time.sleep(5)

pct = lambda xs, q: xs[min(len(xs) - 1, int(len(xs) * q))]
all_lat = sorted(v for vals in lat.values() for v in vals)
spread = sorted(max(v) - min(v) for v in lat.values() if len(v) > 1)
print(f"delivered {len(all_lat)}/{PROBES * MESSAGES}")
if all_lat:
    print(f"latency p50 {pct(all_lat, .5):.1f}  p99 {pct(all_lat, .99):.1f}  max {all_lat[-1]:.1f} ms")
if spread:
    print(f"spread  p50 {pct(spread, .5):.1f}  max {spread[-1]:.1f} ms")

Rode três vezes: sem subscribers do bench (linha de base), com 2.500 e com todos os 10.000. A linha de base mostra a contribuição da rede; o crescimento a partir daí é o custo de fan-out do broker. Depois repita com QOS = 1 e com um RATE maior para descobrir onde a latência começa a subir.

Lendo os resultados

• Entregues < esperados nas sondas em QoS 0 significa que o broker descartou mensagens para algumas conexões — geralmente porque as filas de saída delas estavam cheias. Repita em QoS 1 e verifique os logs do broker.
• A latência p99 cresce com o número de subscribers enquanto a p50 fica estável é normal até certo ponto: o broker escreve nos sockets um após o outro. O que importa é se a p99 continua dentro do seu requisito.
• A dispersão dispara em taxas de publish mais altas significa que o broker (ou a rede dele) não consegue despejar todas as cópias antes da próxima mensagem chegar. Você passou da taxa de fan-out sustentável.
• As máquinas do bench reportam taxas de recebimento menores que o esperado — verifique primeiro CPU e rede delas. Receber 100.000 msg/s é trabalho de verdade para um gerador de carga.
• Alguns subscribers ficam para trás e são desconectados — veja consumidores lentos. Em produção, um dispositivo lento não pode atrasar todos os outros.

Projetando topics para fan-out

O teste costuma terminar em uma conversa de design. Alguns padrões funcionam bem:

1. Broadcast só onde todos realmente precisam. fleet/all/config para mensagens que todo dispositivo deve receber; mantenha-as pequenas.
2. Topics de grupo em vez de um broadcast gigante. fleet/region/eu/config ou fleet/model/x200/firmware reduzem o fan-out por mensagem e permitem rollouts graduais.
3. Retenha o estado atual. Para configuração, publique com a flag retain para que dispositivos que reconectem depois recebam o último valor imediatamente — veja mensagens retidas. Teste se um subscriber atrasado o recebe.
4. Escolha o QoS pela consequência. Atualizações substituíveis em QoS 0; comandos que precisam chegar em QoS 1, dimensionados para que o controle por subscriber do broker continue viável.
5. Evite shared subscriptions aqui. Elas balanceiam carga; não fazem broadcast.

Um checklist curto

1. Calcule a largura de banda de saída para a frota-alvo e a taxa de publish.
2. Distribua os subscribers em máquinas de carga suficientes; aumente os limites de file descriptors.
3. Conecte todos os subscribers primeiro e só depois publique — mantenha os testes de conexão separados.
4. Meça latência e dispersão com sondas, não só com taxas agregadas de recebimento.
5. Repita em QoS 0 e QoS 1, com e sem TLS.
6. Registre CPU, memória e rede de saída do broker junto com os números do cliente.

O fan-out transforma uma mensagem pequena em uma grande rajada de trabalho e tráfego. Faça primeiro a conta de largura de banda, conecte os subscribers antes de publicar e julgue o resultado pelo subscriber mais lento — é ele que os seus usuários vão perceber.

Perguntas frequentes