en
Managed broker Open client
Broker Testing

Build an MQTT Device Simulator in Python

Build an asyncio MQTT device simulator in Python: 1,000 virtual devices with unique client IDs, Last Will, realistic intervals, commands and power-loss faults.

Updated · 10 min read

Build an MQTT Device Simulator in Python

An MQTT device simulator lets you test a broker, a backend or a dashboard against a fleet that does not exist yet. The trick is to make each virtual device behave like a real one: its own connection and unique client ID, a Last Will that marks it offline, telemetry at slightly irregular intervals, commands it reacts to, and the occasional power cut.

This guide builds such a simulator in about 120 lines of Python with asyncio and aiomqtt, then scales it to 1,000 devices and shows how to verify what the broker sees.

What a realistic virtual device does

BehaviourWhy it matters for testing
Own connection, unique client IDExercises per-client sessions, ACLs and connection limits. Duplicate IDs kick each other off.
Staggered bootReal fleets do not connect in the same millisecond — unless power returns after an outage.
Retained online status + Last Will offlineLets dashboards show presence and tests that the broker fires wills.
Jittered publish intervalAvoids perfectly synchronised bursts that real devices never produce.
Heterogeneous fleetSome devices report every 5 s, others every 30 s; payloads differ.
Command topicTests the downlink path: backend → broker → device → acknowledgement.
Power-loss faultsThe device vanishes without DISCONNECT, then reboots and reconnects with backoff.

Topic layout

Each device owns a small subtree. Using a per-run prefix keeps parallel test runs apart:

TopicDirectionRetainedContent
testmqtt/sim/<id>/statusdevice → brokeryesonline, or offline via Last Will
testmqtt/sim/<id>/telemetrydevice → brokernoJSON reading
testmqtt/sim/<id>/cmdbackend → devicenoJSON command, e.g. {"interval": 2}
testmqtt/sim/<id>/cmd/ackdevice → backendnoAcknowledgement

Client IDs follow the pattern sim-<run>-<index>, for example sim-3fa2c1-0042. They are unique, readable in broker logs, and easy to filter. The client ID generator and client ID best practices explain why these properties matter.

The simulator

Install the dependency with pip install aiomqtt (version 2.x, built on paho-mqtt 2.x). Every device is one coroutine with its own 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())

A few details make this simulator trustworthy:

  • random.Random(i) per device — device 42 always has the same interval, starting temperature and crash pattern, so a bug you see once can be reproduced.
  • Retained status with a matching Last Will — the device publishes online after connecting; if it disappears, the broker publishes the retained offline will. Read MQTT Last Will and Testament for the details of when wills fire.
  • Reconnect with exponential backoff and jitter — without it, 1,000 devices that lose the broker at the same moment all come back at the same moment. That pattern, and how to avoid it, is covered in MQTT reconnect storms.
  • Windows — aiomqtt needs the selector event loop there, hence the policy line at the bottom.

Running 1,000 devices

Start the fleet
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

Every ten seconds the simulator prints a line like {'online': 1000, 'published': 48213, 'power_losses': 3, ...}. With the default intervals (5, 10 or 30 s per device), 1,000 devices produce on the order of 100 messages per second — realistic telemetry, not a benchmark. To push harder, shorten the intervals or start more processes with different RUN prefixes; for pure connection pressure, see the MQTT connection stress test.

Verify what the broker sees

Count the devices the broker currently reports as online — the retained status messages are delivered immediately on subscribe:

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

Watch the wills fire in real time by subscribing to the same filter in the online MQTT client and waiting for a power loss: the device's status flips to offline, then back to online after its simulated reboot.

Test the downlink by sending a command to one device and watching for the acknowledgement:

Send a command
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}'

Replace 3fa2c1 with the run ID the simulator printed at startup. After the ack, that device's telemetry should arrive every two seconds.

Extending the simulator

  • Realistic payloads — replace the single temperature value with the sensor model from the MQTT test data generator, including daily cycles, drift and binary encoding.
  • TLS — pass port=8883 and tls_params=aiomqtt.TLSParameters() to aiomqtt.Client to measure the extra cost of TLS handshakes at boot.
  • MQTT 5 — add protocol=aiomqtt.ProtocolVersion.V5 and session expiry properties to test persistent sessions and message expiry.
  • Device types — run several coroutine functions (thermostat, meter, gateway) with different topics and rates from the same main().
  • Mass reconnect — set CRASH_PROB to zero and restart the broker during a run to see how the whole fleet recovers.

Watch simulated devices come online

A simulator turns "we think it scales" into an experiment you can rerun every week. Give each virtual device a real identity, make it fail the way hardware fails, and verify the result on the broker — then the first thousand real devices are just a repeat of a test you have already passed.

Frequently asked questions

How many simulated MQTT devices can one machine run?

With asyncio, one Python process comfortably runs hundreds to a few thousand devices at typical telemetry rates, because each device is a coroutine rather than a thread. The practical limits are the file-descriptor limit (ulimit -n), the broker's connection limits and CPU spent on JSON and TLS. Beyond that, run several processes or machines.

Should every simulated device have its own MQTT connection?

Yes, if you want to test the broker the way a real fleet uses it. One connection per device exercises connection handling, keep-alive, sessions, Last Will and ACLs per client. Publishing for many devices over one connection only tests throughput.

How do I simulate a device losing power?

Close the TCP connection without sending DISCONNECT. The broker then treats it as an unexpected disconnect and publishes the device's Last Will. A clean disconnect, by contrast, discards the will, so it does not test your offline detection.

Keep reading

Broker Testing

Generate Realistic MQTT Test Data (Sensors, JSON, Binary)

Generate realistic MQTT test data: Python and Node.js scripts for sensor telemetry with daily cycles, noise and drift, encoded as JSON or compact binary.

9 min readTutorials

Async MQTT in Python with aiomqtt

Use aiomqtt for async MQTT in Python: connect, subscribe with async for, publish from tasks, reconnect after errors and run it inside a FastAPI app.

8 min readBroker Testing

MQTT Connection Stress Test: How Many Clients Can a Broker Handle?

MQTT connection stress test: open tens of thousands of clients, fix ulimit and ephemeral port limits, use multiple source IPs and simulate reconnect storms.

9 min read