Locust is a Python load-testing framework with a live web dashboard, distributed workers and test scenarios written as plain code. It speaks HTTP by default, but its User class can drive any protocol — so MQTT load testing with Locust comes down to wrapping paho-mqtt in a user class and reporting each connect and publish to Locust's statistics.
This guide gives you a complete locustfile.py: a reusable MQTT base user, sensor users that publish telemetry, a dashboard user that measures end-to-end latency, and the commands to run it from the browser, headless in CI and across several machines.
Why Locust for MQTT?
| Tool | Scenario language | MQTT support | Strong point |
|---|---|---|---|
| Locust | Python | Custom User class (or locust-plugins) | Realistic behaviour in plain Python, live web UI |
| k6 | JavaScript | Extension (xk6-mqtt) | Scripted tests and thresholds in CI |
| JMeter | GUI / XML test plans | Plugin | Teams already standardised on JMeter |
| emqtt-bench | CLI flags | Native | Raw connection and throughput numbers |
Locust is the right pick when a "user" must behave like a device: connect, publish on a schedule, react to messages, reconnect — and you want to write that logic in Python. For a quick throughput ceiling, a CLI benchmark is simpler; see MQTT broker load testing.
How the integration works
Every Locust user is a lightweight greenlet (Locust runs on gevent). Our MQTT user:
- Creates one paho-mqtt client per Locust user, so 1,000 users means 1,000 MQTT connections.
- Measures connect time from
connect()to CONNACK. - Measures publish time from
publish()to PUBACK for QoS 1 (QoS 0 is reported as soon as it is written). - Reports each result with
environment.events.request.fire(...), which is exactly what Locust's HTTP client does internally — so the dashboard, CSV export and failure tab all work unchanged.
The locustfile
import json, os, random, threading, time, uuid
import paho.mqtt.client as mqtt
from locust import User, between, constant, task
from locust.exception import StopUser
HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", 1883))
PREFIX = "testmqtt/locust"
class MqttUser(User):
abstract = True
subscriptions: list[tuple[str, int]] = []
def on_start(self):
self.client_id = f"locust-{uuid.uuid4().hex[:10]}"
self.pending, self.lock = {}, threading.Lock()
self.connected, self.connect_rc = threading.Event(), None
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
if os.environ.get("MQTT_USER"):
c.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
c.on_connect, c.on_publish, c.on_message = self._on_connect, self._on_publish, self.on_message
self.client = c
start = time.perf_counter()
try:
c.connect(HOST, PORT, keepalive=60)
c.loop_start()
if not self.connected.wait(10):
raise TimeoutError("no CONNACK within 10 s")
if self.connect_rc.is_failure:
raise ConnectionError(f"CONNACK: {self.connect_rc}")
self._report("CONNECT", "connect", start, 0)
except Exception as e:
self._report("CONNECT", "connect", start, 0, e)
c.loop_stop()
raise StopUser()
def on_stop(self):
with self.lock:
leftovers, self.pending = list(self.pending.values()), {}
for start, name, size in leftovers:
self._report("PUBLISH", name, start, size, TimeoutError("no PUBACK before stop"))
self.client.disconnect()
self.client.loop_stop()
# --- reporting -------------------------------------------------------
def _report(self, kind, name, start, size, exc=None):
self.environment.events.request.fire(
request_type=kind, name=name, response_time=(time.perf_counter() - start) * 1000,
response_length=size, exception=exc, context={"client_id": self.client_id})
# --- paho callbacks (run in the network loop) -----------------------
def _on_connect(self, client, userdata, flags, reason_code, properties):
self.connect_rc = reason_code
for topic, qos in self.subscriptions: # also restores them after a reconnect
client.subscribe(topic, qos)
self.connected.set()
def _on_publish(self, client, userdata, mid, reason_code, properties):
with self.lock:
item = self.pending.pop(mid, None)
if item:
start, name, size = item
exc = Exception(f"PUBACK: {reason_code}") if reason_code.is_failure else None
self._report("PUBLISH", name, start, size, exc)
def on_message(self, client, userdata, msg):
pass
# --- API for tasks ---------------------------------------------------
def publish(self, topic, payload, qos=1, name=None):
name, start = name or topic, time.perf_counter()
with self.lock:
info = self.client.publish(topic, payload, qos=qos)
if info.rc == mqtt.MQTT_ERR_SUCCESS and qos > 0:
self.pending[info.mid] = (start, name, len(payload))
return
exc = None if info.rc == mqtt.MQTT_ERR_SUCCESS else Exception(mqtt.error_string(info.rc))
self._report("PUBLISH", name, start, len(payload), exc)
class Sensor(MqttUser):
weight = 20
wait_time = between(4, 6) # one reading every ~5 s
@task(10)
def telemetry(self):
payload = json.dumps({"ts": time.time(), "temp": round(random.gauss(21, 0.5), 2)})
self.publish(f"{PREFIX}/{self.client_id}/telemetry", payload, qos=1, name="telemetry")
@task(1)
def heartbeat(self):
self.publish(f"{PREFIX}/{self.client_id}/status", "ok", qos=0, name="status")
class Dashboard(MqttUser):
weight = 1
wait_time = constant(1)
subscriptions = [(f"{PREFIX}/+/telemetry", 1)]
def on_message(self, client, userdata, msg):
sent = json.loads(msg.payload)["ts"]
self.environment.events.request.fire(
request_type="RECEIVE", name="telemetry end-to-end",
response_time=(time.time() - sent) * 1000, response_length=len(msg.payload),
exception=None, context={})
@task
def idle(self):
pass # messages arrive via on_messageSome details that matter:
name="telemetry"groups every sensor's publish into one row. Without it, each client's unique topic becomes its own line in the statistics table and the dashboard becomes unreadable.- The lock around
publish()andpendingprevents a race where a very fast PUBACK is processed before the message ID has been recorded. - Weights — with
weight = 20andweight = 1, a test with 210 users starts about 200 sensors and 10 dashboards. Every dashboard receives every sensor's telemetry, so dashboards also test fan-out. - End-to-end latency compares
time.time()on two clients. That is accurate on a single machine; across machines, the clocks must be synchronised with NTP or the numbers include clock skew.
Running the test
pip install locust "paho-mqtt>=2"
export MQTT_HOST=your-broker-host # your own broker instance, not the shared endpoint
export MQTT_USER=username MQTT_PASS=password
ulimit -n 8192 # one socket per user
locust -f locustfile.pyOpen http://localhost:8089, enter the number of users and the spawn rate (users started per second), and start. The Statistics tab shows one row per name — connect, telemetry, status, telemetry end-to-end — with request rate, failures and percentiles. Charts plots total requests per second, response times and user count live; Failures groups errors such as The client is not currently connected. or no CONNACK within 10 s.
For CI or long unattended runs, use headless mode and keep the reports:
locust -f locustfile.py --headless \
--users 2000 --spawn-rate 50 --run-time 15m \
--csv results/mqtt --html results/mqtt-report.html--spawn-rate is your connection rate. Keep it at a value your real fleet could plausibly produce; 2,000 users at 50 per second ramps in 40 seconds. A very high spawn rate turns a load test into a reconnect-storm test — useful, but a different question.
Before scaling up, run with a handful of users and watch the topics in the online MQTT client (testmqtt/locust/#) to confirm payloads and topic names look right. Smoke tests like that are fine on the shared broker.connectmq.com endpoint, but it rate-limits traffic, so set MQTT_HOST to your own broker instance for the actual load test.
Scaling out: workers and processes
One Locust process uses one CPU core. When the load generator saturates, Locust logs CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements! — at that point, response times reflect your laptop, not the broker. Add processes:
# all cores on one machine (Linux/macOS)
locust -f locustfile.py --processes -1
# or a master plus workers on several machines
locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host 10.0.0.5Workers share the same locustfile; the master aggregates statistics into one dashboard. Spread workers across source IPs if you go beyond tens of thousands of connections — ephemeral port and file-descriptor limits on a single host are a common false ceiling.
Reading Locust results for MQTT
| Row | What it measures | What to look for |
|---|---|---|
CONNECT connect | TCP + MQTT handshake until CONNACK | Rising times during ramp-up mean the broker or auth backend struggles with connection rate |
PUBLISH telemetry (QoS 1) | Publish until PUBACK | p95/p99 creeping up at constant users means queues are building |
PUBLISH status (QoS 0) | Time to hand the packet to the socket | Should stay near zero; failures mean the client was disconnected |
RECEIVE telemetry end-to-end | Publisher timestamp to subscriber delivery | The latency your applications actually feel |
Compare the RPS of PUBLISH telemetry with RECEIVE divided by the number of dashboards: if receiving falls behind publishing, messages are queuing or being dropped. QoS changes the packet flow and therefore the numbers — MQTT QoS levels explains why QoS 1 latency includes a full round trip. For a structured way to turn percentiles into a capacity decision, see interpreting MQTT benchmark results.
Variations worth adding
- TLS — call
c.tls_set()beforeconnect()and setMQTT_PORT=8883to include TLS handshakes in connect time. - MQTT 5 — pass
protocol=mqtt.MQTTv5tomqtt.Clientto test MQTT 5 brokers; PUBACK reason codes then show up as failures in the dashboard. - Load shapes — a
LoadTestShapeclass can ramp in steps or simulate a morning peak automatically. - Commands — add a backend user that publishes to
testmqtt/locust/<id>/cmdand have sensors acknowledge, to load-test the downlink path too.
Locust's strength for MQTT is that a user is just Python: realistic device behaviour, a dashboard your whole team can read, and the same file running on a laptop, in CI or across a fleet of workers. Start small, verify the topics, then ramp up and let the percentiles tell you where the limits are.
Frequently asked questions
Does Locust support MQTT out of the box?
No. Locust ships with an HTTP user, but any protocol can be tested by writing a User class that calls a client library and reports each operation through the request event. For MQTT that means wrapping paho-mqtt; the community locust-plugins package also contains an MQTT user you can start from.
What does response time mean for an MQTT publish in Locust?
Whatever you choose to measure. In this guide it is the time from calling publish until the broker's PUBACK arrives for QoS 1, which is the broker's acknowledgement latency. QoS 0 has no acknowledgement, so end-to-end delivery latency is measured separately by a subscribing user.
How many MQTT users can one Locust process simulate?
It depends on message rate and payload handling. Each Locust process uses one CPU core, so watch the CPU warning Locust prints; when it appears, add worker processes with --processes or run workers on more machines instead of adding users to one process.
