Ferramentas de Teste de Carga MQTT: emqtt-bench, MQTTX Bench e Locust Comparados

As ferramentas de teste de carga MQTT dizem se o seu sistema aguenta quando milhares de dispositivos se conectam ao mesmo tempo (pense nos medidores de uma distribuidora de energia voltando após uma queda), quando um bug de firmware inunda um topic ou quando um assinante do backend fica para trás. Este guia compara emqtt-bench, MQTTX Bench e Locust, mostra os comandos de cada um e acrescenta uma pequena sonda em Python que mede a latência de ponta a ponta e a perda de mensagens. Você também vai aprender quais métricas importam e como testar com responsabilidade, sempre contra o seu próprio broker privado, nunca um público compartilhado.

Primeira regra: teste o seu próprio broker

Escolher entre as ferramentas de teste de carga MQTT disponíveis é a parte fácil; o que importa é apontá-las para o alvo certo. Testes de carga geram tráfego que parece um ataque. Execute-os apenas contra infraestrutura que é sua ou que você tem permissão explícita para testar.

• Não faça teste de carga no servidor MQTT público. broker.connectmq.com, o servidor MQTT público (gratuito) do ConnectMQ, é um serviço compartilhado para testes e protótipos, com limites rígidos por cliente (10 mensagens por segundo por cliente, 10 assinaturas por cliente, mensagens de 64 KB). Um teste de carga ali mede esses limites, não o seu sistema, e piora o serviço para todos os outros.
• Use um broker privado dimensionado para o teste. No ConnectMQ, crie um broker dedicado ao teste de carga para que os dispositivos de produção não sejam afetados.
• Conheça as cotas do seu plano antes de começar. O plano gratuito Serverless permite até 25 conexões e QoS até 1; o Serverless pago permite até 1.000 conexões, QoS 2 e mensagens de 1 MB; o Dedicated roda em um cluster isolado. Ultrapassar uma cota não é falha do broker: clientes MQTT 5 recebem o reason code 0x97 (quota exceeded), que o seu teste deve registrar separadamente dos erros reais.
• Confira as regras de client ID. O ConnectMQ vincula os client IDs às credenciais que emite, enquanto as ferramentas de carga geram IDs a partir de um prefixo e um contador. Confirme no portal quais client IDs as suas credenciais de teste permitem e configure a opção de client ID da ferramenta de acordo, ou todas as conexões serão rejeitadas como não autorizadas.
• Atenção às camadas de proteção. Brokers MQTT gerenciados aplicam proteção contra flood e bloqueiam IPs após repetidas falhas de login. Um teste mal configurado, com credenciais erradas, pode fazer o IP da sua máquina de teste ser bloqueado, então valide as credenciais primeiro com um único cliente.

O que medir

A vazão sozinha é um número enganoso. Colete estes dados e defina os critérios de aprovação antes de rodar qualquer coisa:

Sempre monitore o gerador de carga. Se a CPU dele estiver em 100 por cento, os números de latência refletem o seu notebook, não o broker. Divida testes grandes entre várias máquinas.

emqtt-bench

O emqtt-bench é uma ferramenta leve em Erlang do projeto EMQX, projetada para abrir muitas conexões com pouquíssima memória por cliente. Ela tem três modos: conn, pub e sub.

# Ramp up 500 connections, one every 10 ms
emqtt_bench conn -h YOUR_BROKER_HOST -p 1883 -c 500 -i 10 \
  -u YOUR_USERNAME -P YOUR_PASSWORD

# 50 subscribers on a wildcard, QoS 1
emqtt_bench sub -h YOUR_BROKER_HOST -p 1883 -c 50 -i 10 \
  -t 'loadtest/#' -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

# 200 publishers, each sending a 256-byte message every 1000 ms to its own topic
emqtt_bench pub -h YOUR_BROKER_HOST -p 1883 -c 200 -i 10 -I 1000 \
  -t 'loadtest/device/%i' -s 256 -q 1 -u YOUR_USERNAME -P YOUR_PASSWORD

Flags principais: -c quantidade de clientes, -i intervalo entre novas conexões em milissegundos, -I intervalo entre publicações por cliente, -t topic com os placeholders %i (número sequencial do cliente) e %c (client ID), -s tamanho do payload, -q QoS. Adicione --ssl e a porta 8883 para testar TLS, o que você deveria fazer, porque os handshakes TLS costumam ser o custo real durante tempestades de reconexão. A ferramenta imprime as taxas a cada poucos segundos.

Pontos fortes: muito eficiente por conexão, ideal para testes de quantidade de conexões e de tempestades de reconexão. Pontos fracos: os payloads são sintéticos e a medição de latência é limitada, então combine-a com uma sonda (abaixo).

MQTTX Bench

O MQTTX CLI inclui um comando bench com os mesmos três modos e uma saída mais amigável. Se a sua equipe já usa o MQTTX, é o começo mais fácil.

mqttx bench conn -c 200 -i 10 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench sub -c 20 -t 'loadtest/#' -q 1 -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

mqttx bench pub -c 100 -i 10 -im 1000 -t 'loadtest/device/%i' -q 1 \
  -m '{"temp": 21.5}' -h YOUR_BROKER_HOST -p 8883 -l mqtts \
  -u YOUR_USERNAME -P YOUR_PASSWORD

Aqui, -i é o intervalo entre conexões e -im o intervalo de mensagens por cliente. Como roda em Node.js, cada cliente custa mais memória do que no emqtt-bench, então quantidades muito grandes de conexões exigem mais máquinas geradoras. Consulte mqttx bench pub --help na sua versão, pois as opções evoluem.

Locust

O Locust é um framework de teste de carga em Python. Ele não tem suporte nativo a MQTT, mas você pode escrever uma classe User em torno do paho-mqtt e enviar os resultados para a interface web do Locust. Isso o torna a opção mais flexível: payloads realistas, comportamento por dispositivo, formatos de rampa e asserções são simplesmente Python.

# locustfile.py  (pip install locust paho-mqtt)
import json
import os
import random
import time
import uuid

import paho.mqtt.client as mqtt
from locust import User, between, events, task

HOST = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST")
USER = os.environ.get("MQTT_USER", "YOUR_USERNAME")
PASS = os.environ.get("MQTT_PASS", "YOUR_PASSWORD")


class MqttDevice(User):
    wait_time = between(1, 2)

    def on_start(self):
        self.device_id = uuid.uuid4().hex[:8]
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"lt-{self.device_id}",
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(USER, PASS)
        self.client.tls_set()
        start = time.perf_counter()
        try:
            self.client.connect(HOST, 8883, keepalive=30)
            self.client.loop_start()
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=None, context={})
        except Exception as exc:
            events.request.fire(request_type="MQTT", name="connect",
                                response_time=(time.perf_counter() - start) * 1000,
                                response_length=0, exception=exc, context={})
            raise

    @task
    def publish_telemetry(self):
        payload = json.dumps({"id": self.device_id,
                              "temp": round(random.uniform(18, 26), 2),
                              "ts": time.time()})
        start = time.perf_counter()
        info = self.client.publish(f"loadtest/device/{self.device_id}", payload, qos=1)
        exc = None
        try:
            info.wait_for_publish(timeout=5)  # aguarda o PUBACK no QoS 1
            if not info.is_published():
                exc = TimeoutError("no PUBACK within 5 s")
        except Exception as e:
            exc = e
        events.request.fire(request_type="MQTT", name="publish qos1",
                            response_time=(time.perf_counter() - start) * 1000,
                            response_length=len(payload), exception=exc, context={})

    def on_stop(self):
        self.client.loop_stop()
        self.client.disconnect()

Execute com locust -f locustfile.py --headless -u 100 -r 5 --run-time 10m, onde -u é o número de dispositivos simulados e -r a taxa de criação por segundo. Cada dispositivo roda sua própria thread de rede do paho, então o Locust é mais pesado por conexão do que o emqtt-bench; use-o para testes ricos em comportamento com centenas de dispositivos, e não para contagens brutas de conexões. A medição de tempo da publicação aqui mede a ida e volta do PUBACK, um bom indicador da capacidade de resposta do broker.

Uma sonda de latência e perda em Python

Nenhuma das ferramentas acima entrega, por padrão, números confiáveis de latência de ponta a ponta e de perda. Rode esta sonda junto com qualquer teste de carga: ela publica mensagens sequenciadas e com timestamp e mede o que chega.

# probe.py  (pip install paho-mqtt)
import json, os, statistics, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = os.environ.get("MQTT_HOST", "YOUR_BROKER_HOST"), 8883
USER, PASS = os.environ.get("MQTT_USER", "YOUR_USERNAME"), os.environ.get("MQTT_PASS", "YOUR_PASSWORD")
TOPIC = f"loadtest/probe/{uuid.uuid4().hex[:6]}"
COUNT, INTERVAL = 600, 0.5

latencies, seen = [], set()

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    latencies.append((time.time() - data["ts"]) * 1000)
    seen.add(data["seq"])

def make(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{name}-{uuid.uuid4().hex[:6]}")
    c.username_pw_set(USER, PASS)
    c.tls_set()
    c.connect(HOST, PORT, keepalive=30)
    c.loop_start()
    return c

sub = make("sub")
sub.on_message = on_message
sub.subscribe(TOPIC, qos=1)
time.sleep(1)
pub = make("pub")

for seq in range(COUNT):
    pub.publish(TOPIC, json.dumps({"seq": seq, "ts": time.time()}), qos=1)
    time.sleep(INTERVAL)

time.sleep(5)
q = statistics.quantiles(latencies, n=100)
print(f"received {len(seen)}/{COUNT}, lost {COUNT - len(seen)}")
print(f"latency ms p50={q[49]:.1f} p95={q[94]:.1f} p99={q[98]:.1f} max={max(latencies):.1f}")

Publicador e assinante rodam na mesma máquina, então a diferença entre relógios não distorce a latência. Rode a sonda uma vez com o broker ocioso para ter uma linha de base e depois durante cada estágio de carga; a diferença é o custo da carga. Se as suas credenciais permitem apenas client IDs específicos, substitua os IDs gerados pelos que foram emitidos para você.

Resumo comparativo

Um plano de teste responsável

1. Escreva o objetivo: por exemplo, "2.000 dispositivos reconectam em até 2 minutos; a latência p95 fica abaixo de um limite acordado".
2. Valide com um único cliente: credenciais, TLS, estrutura de topics, regras de client ID.
3. Linha de base: rode a sonda com o broker ocioso.
4. Aumente em estágios: 10, 25, 50 e 100 por cento da meta, mantendo cada estágio por vários minutos.
5. Teste os casos difíceis: todos reconectando ao mesmo tempo, um assinante que para de ler, payloads grandes demais.
6. Registre as respostas de cota separadamente: 0x97 significa que você atingiu um limite do plano, o que é informação, não uma pane.
7. Faça a limpeza: apague as retained messages de teste e desconecte todos os clientes.

Se os resultados mostrarem que você precisa de mais conexões ou de mais isolamento do que o seu plano atual oferece, esse é o sinal para subir de nível, por exemplo de Serverless para Dedicated. A visão geral do broker MQTT gerenciado e o guia para escolher um broker MQTT na nuvem explicam os prós e contras. Para testes funcionais antes do teste de carga, veja como testar clientes MQTT, e para gerar tráfego realista em baixo volume use um simulador de dispositivos MQTT. O básico de linha de comando está em ferramentas CLI para MQTT, e o guia de bibliotecas cliente MQTT ajuda se você quiser construir um gerador de carga próprio em outra linguagem.

Comece a usar o ConnectMQ

Crie um broker de testes dedicado em app.connectmq.com no plano gratuito Serverless, sem cartão de crédito, e valide seus scripts de teste em pequena escala. Quando estiver pronto para execuções maiores, faça o upgrade da mesma configuração de broker e mantenha o dashboard do portal aberto para acompanhar as conexões enquanto o teste roda.

Perguntas frequentes

Posso fazer teste de carga no servidor MQTT público?

Não. Servidores MQTT públicos são compartilhados e têm limite de taxa por cliente, então os resultados não significam nada e outros usuários são afetados. Use o seu próprio broker privado.

Qual é a melhor ferramenta de teste de carga MQTT?

emqtt-bench para grandes quantidades de conexões, MQTTX Bench para testes rápidos e legíveis, e Locust quando você precisa de comportamento realista de dispositivos. Muitas equipes usam o emqtt-bench para escala mais uma pequena sonda para latência.

Por que os clientes do meu teste de carga são rejeitados com quota exceeded?

Você atingiu um limite do seu plano, como o número máximo de conexões. Clientes MQTT 5 recebem o reason code 0x97; reduza a carga ou migre para um plano com limites maiores.

Os testes de carga devem usar TLS?

Sim, se a produção usa TLS. Os handshakes acrescentam um trabalho de CPU significativo dos dois lados durante tempestades de reconexão, e testar sem TLS esconde esse custo.