<- Test Directory

MQTT Load Testing

How to customize LoadForge tests to test stress test MQTT servers


MQTT is a lightweight, publish-subscribe network protocol that transports messages between devices.

By default, LoadForge tests plain HTTP/HTTPS sites. However, as you can use python for test definitions it is easy to expand beyond that.

The below example allows you to load test an MQTT service. This use case is not officially supported, but you can see the example script below from the locust community.


import os
import ssl
import time

from locust import task, TaskSet
from locust.user.wait_time import between
from locust_plugins.users import MqttUser

tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS)

class MyUser(MqttUser):
    host = "localhost"
    port = 8883
    tls_context = tls_context
    # We could uncomment below to use the WebSockets transport
    # transport = "websockets"
    # ws_path = "/mqtt/custom/path"

    # We'll probably want to throttle our publishing a bit: let's limit it to
    # 10-100 messages per second.
    wait_time = between(0.01, 0.1)

    class MyTasks(TaskSet):
        # Sleep for a while to allow the client time to connect.
        def on_start(self):

        def say_hello(self):
            self.client.publish("hello/locust", b"hello world")    

This guide is part of the LoadForge Directory, an index of locustfile's for use with LoadForge website and API load tests. We also provide a wizard to generate tests, and onboarding assistance for clients. Contact us should you have any questions.

LoadForge provides load testing and stress tests for websites, APIs and servers. It uses your cloud account to rapidly scale large numbers of simulated users to load test your website, store, API, or application for cheap - just cents per test!

For more help on Tests please see our official documentation. Logged in users can also use our wizard to generate a locustfile, or you can record your browser steps.

Ready to run your test?
Start your first test within minutes.