Testes de brokers

Crie um simulador de dispositivos MQTT em Python

Crie um simulador de dispositivos MQTT com asyncio em Python: 1.000 dispositivos virtuais com client IDs únicos, Last Will, intervalos realistas e falhas.

Atualizado · 10 min de leitura

Crie um simulador de dispositivos MQTT em Python

Um simulador de dispositivos MQTT permite testar um broker, um backend ou um dashboard contra uma frota que ainda não existe. O segredo é fazer cada dispositivo virtual se comportar como um real: conexão própria e client ID único, um Last Will que o marca como offline, telemetria em intervalos levemente irregulares, comandos aos quais ele responde e uma queda de energia de vez em quando.

Este guia monta esse simulador em cerca de 120 linhas de Python com asyncio e aiomqtt, escala para 1.000 dispositivos e mostra como verificar o que o broker enxerga.

O que um dispositivo virtual realista faz

ComportamentoPor que importa nos testes
Conexão própria, client ID únicoExercita sessões por cliente, ACLs e limites de conexão. IDs duplicados derrubam um ao outro.
Boot escalonadoFrotas reais não conectam no mesmo milissegundo — a não ser quando a energia volta depois de um apagão.
Status online retido + Last Will offlinePermite que dashboards mostrem presença e testa se o broker dispara os wills.
Intervalo de publicação com jitterEvita rajadas perfeitamente sincronizadas que dispositivos reais nunca produzem.
Frota heterogêneaAlguns dispositivos reportam a cada 5 s, outros a cada 30 s; os payloads variam.
Topic de comandosTesta o caminho de descida: backend → broker → dispositivo → acknowledgement.
Falhas por queda de energiaO dispositivo some sem DISCONNECT, depois reinicia e reconecta com backoff.

Estrutura de topics

Cada dispositivo é dono de uma pequena subárvore. Um prefixo por execução mantém separados os testes que rodam em paralelo:

TopicDireçãoRetidoConteúdo
testmqtt/sim/<id>/statusdispositivo → brokersimonline, ou offline via Last Will
testmqtt/sim/<id>/telemetrydispositivo → brokernãoLeitura em JSON
testmqtt/sim/<id>/cmdbackend → dispositivonãoComando JSON, ex. {"interval": 2}
testmqtt/sim/<id>/cmd/ackdispositivo → backendnãoConfirmação

Os client IDs seguem o padrão sim-<run>-<index>, por exemplo sim-3fa2c1-0042. São únicos, legíveis nos logs do broker e fáceis de filtrar. O gerador de client ID e as boas práticas de client ID explicam por que essas propriedades importam.

O simulador

Instale a dependência com pip install aiomqtt (versão 2.x, baseada no paho-mqtt 2.x). Cada dispositivo é uma corrotina com seu próprio aiomqtt.Client:

simulator.py
import asyncio, json, os, random, socket, sys, time, uuid
from collections import Counter
import aiomqtt

HOST, PORT = os.environ.get("MQTT_HOST", "broker.connectmq.com"), 1883
USER, PASS = os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS")
N_DEVICES = int(os.environ.get("DEVICES", 100))
RAMP_S = 60              # spread device boot over one minute
CRASH_PROB = 0.002       # chance of a power loss after each publish
RUN = uuid.uuid4().hex[:6]
stats = Counter()

class PowerLoss(Exception):
    pass

async def commands(client, base, state):
    try:
        async for msg in client.messages:
            try:
                cmd = dict(json.loads(msg.payload))
                if "interval" in cmd:
                    state["interval"] = max(1.0, float(cmd["interval"]))
            except (ValueError, TypeError):
                continue                                   # ignore malformed commands
            await client.publish(f"{base}/cmd/ack", json.dumps({"ok": True, **cmd}), qos=1)
            stats["commands"] += 1
    except aiomqtt.MqttError:
        pass                                               # connection lost; device() reconnects

async def device(i: int):
    dev_id = f"sim-{RUN}-{i:04d}"
    base = f"testmqtt/sim/{dev_id}"
    rng = random.Random(i)
    state = {"interval": rng.choice([5, 10, 30])}
    temp, boots, attempt = rng.uniform(19, 24), 0, 0
    await asyncio.sleep(rng.uniform(0, RAMP_S))          # staggered boot

    while True:
        try:
            will = aiomqtt.Will(f"{base}/status", b"offline", qos=1, retain=True)
            async with aiomqtt.Client(HOST, PORT, identifier=dev_id, username=USER,
                                      password=PASS, keepalive=30, will=will) as client:
                attempt, boots, started = 0, boots + 1, time.monotonic()
                stats["online"] += 1
                try:
                    await client.publish(f"{base}/status", b"online", qos=1, retain=True)
                    await client.subscribe(f"{base}/cmd", qos=1)
                    listener = asyncio.create_task(commands(client, base, state))
                    try:
                        while True:
                            temp += rng.gauss(0, 0.05)
                            reading = {"temp": round(temp, 2), "boots": boots,
                                       "uptime": int(time.monotonic() - started), "fw": "1.4.2"}
                            await client.publish(f"{base}/telemetry", json.dumps(reading), qos=1)
                            stats["published"] += 1
                            if rng.random() < CRASH_PROB:
                                # power loss: drop TCP without DISCONNECT so the broker fires the will
                                # (aiomqtt has no public API for this, so reach the paho client)
                                client._client.socket().shutdown(socket.SHUT_RDWR)
                                await asyncio.sleep(1)
                                raise PowerLoss
                            await asyncio.sleep(state["interval"] * rng.uniform(0.9, 1.1))
                    finally:
                        listener.cancel()
                finally:
                    stats["online"] -= 1
        except PowerLoss:
            stats["power_losses"] += 1
            await asyncio.sleep(rng.uniform(5, 30))       # time to reboot
        except aiomqtt.MqttError:
            stats["conn_errors"] += 1
            attempt += 1
            await asyncio.sleep(min(60, 2 ** attempt) * rng.uniform(0.5, 1.0))

async def report():
    while True:
        await asyncio.sleep(10)
        print(time.strftime("%H:%M:%S"), dict(stats), flush=True)

async def main():
    print(f"run {RUN}: starting {N_DEVICES} devices under testmqtt/sim/sim-{RUN}-*")
    await asyncio.gather(report(), *(device(i) for i in range(N_DEVICES)))

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())

Alguns detalhes tornam este simulador confiável:

  • random.Random(i) por dispositivo — o dispositivo 42 sempre tem o mesmo intervalo, a mesma temperatura inicial e o mesmo padrão de falhas, então um bug visto uma vez pode ser reproduzido.
  • Status retido com o Last Will correspondente — o dispositivo publica online depois de conectar; se ele sumir, o broker publica o will retido offline. Leia Last Will and Testament no MQTT para saber exatamente quando os wills disparam.
  • Reconexão com backoff exponencial e jitter — sem isso, 1.000 dispositivos que perdem o broker no mesmo instante voltam todos no mesmo instante. Esse padrão, e como evitá-lo, está em tempestades de reconexão MQTT.
  • Windows — lá o aiomqtt precisa do event loop baseado em selector, por isso a linha da policy no final.

Rodando 1.000 dispositivos

Iniciar a frota
ulimit -n 4096                     # one socket per device, plus headroom
export MQTT_HOST=your-broker-host    # your own broker instance for the full fleet
export MQTT_USER=username MQTT_PASS=password
DEVICES=1000 python simulator.py

A cada dez segundos o simulador imprime uma linha como {'online': 1000, 'published': 48213, 'power_losses': 3, ...}. Com os intervalos padrão (5, 10 ou 30 s por dispositivo), 1.000 dispositivos geram algo na ordem de 100 mensagens por segundo — telemetria realista, não um benchmark. Para apertar mais, encurte os intervalos ou inicie mais processos com prefixos RUN diferentes; para pressão pura de conexões, veja o teste de estresse de conexões MQTT.

Verifique o que o broker enxerga

Conte os dispositivos que o broker considera online agora — as mensagens de status retidas são entregues assim que você assina:

Contar dispositivos online
mosquitto_sub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/+/status' -v -W 5 | grep -c ' online$'

Veja os wills dispararem em tempo real assinando o mesmo filtro no cliente MQTT online e esperando uma queda de energia: o status do dispositivo muda para offline e volta para online depois do reboot simulado.

Teste o caminho de descida enviando um comando para um dispositivo e aguardando a confirmação:

Enviar um comando
mosquitto_sub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/sim-3fa2c1-0042/cmd/ack' -C 1 -v &
mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/sim-3fa2c1-0042/cmd' -q 1 -m '{"interval": 2}'

Troque 3fa2c1 pelo ID de execução que o simulador mostrou ao iniciar. Depois da confirmação, a telemetria desse dispositivo deve chegar a cada dois segundos.

Estendendo o simulador

  • Payloads realistas — troque o valor único de temperatura pelo modelo de sensor do gerador de dados de teste MQTT, com ciclo diário, deriva e codificação binária.
  • TLS — passe port=8883 e tls_params=aiomqtt.TLSParameters() para o aiomqtt.Client para medir o custo extra dos handshakes TLS no boot.
  • MQTT 5 — adicione protocol=aiomqtt.ProtocolVersion.V5 e propriedades de session expiry para testar sessões persistentes e expiração de mensagens.
  • Tipos de dispositivo — rode várias funções corrotina (thermostat, meter, gateway) com topics e taxas diferentes a partir do mesmo main().
  • Reconexão em massa — coloque CRASH_PROB em zero e reinicie o broker durante uma execução para ver como a frota inteira se recupera.

Veja os dispositivos simulados ficarem online

Um simulador transforma "achamos que escala" num experimento que você pode repetir toda semana. Dê a cada dispositivo virtual uma identidade real, faça-o falhar como o hardware falha e verifique o resultado no broker — assim os primeiros mil dispositivos reais serão só a repetição de um teste em que você já passou.

Perguntas frequentes

Quantos dispositivos MQTT simulados uma máquina consegue rodar?

Com asyncio, um processo Python roda tranquilamente de centenas a alguns milhares de dispositivos em taxas típicas de telemetria, porque cada dispositivo é uma corrotina e não uma thread. Os limites práticos são o limite de descritores de arquivo (ulimit -n), os limites de conexão do broker e a CPU gasta com JSON e TLS. Acima disso, use vários processos ou máquinas.

Cada dispositivo simulado deve ter sua própria conexão MQTT?

Sim, se você quer testar o broker do jeito que uma frota real o usa. Uma conexão por dispositivo exercita o tratamento de conexões, keep-alive, sessões, Last Will e ACLs por cliente. Publicar por muitos dispositivos numa única conexão só testa throughput.

Como simular um dispositivo perdendo energia?

Feche a conexão TCP sem enviar DISCONNECT. O broker trata isso como uma desconexão inesperada e publica o Last Will do dispositivo. Uma desconexão limpa, por outro lado, descarta o will e por isso não testa sua detecção de dispositivos offline.