MQTT Fan-In Test: Thousands of Devices, One Backend

Fan-in is the everyday shape of IoT: thousands of devices each publish a little telemetry, and one backend subscribes to devices/+/telemetry to store or process it all. Individually the devices are quiet; together they send everything through one subscription, one TCP connection and one process. A fan-in test finds out how many devices that consumer can absorb before a backlog builds up — and how to scale it when it cannot.

This guide simulates a fleet with emqtt-bench, measures backlog and lag on the consumer side with a Python script, and shows how shared subscriptions spread the load. The reverse pattern is covered in the fan-out test.

Where fan-in bottlenecks hide

The broker usually handles fan-in ingress comfortably — routing many small messages to one subscriber is cheap. The trouble is downstream:

The last two are broker-side limits on a single session — for example Mosquitto's max_queued_messages and max_inflight_messages. When the consumer falls behind, the broker queues messages for it, and once the queue is full it starts dropping or disconnecting, depending on the broker. See slow consumers for how different brokers behave.

Step 1: simulate the fleet

emqtt-bench can play thousands of devices from one machine. Give each client its own topic with %i (the client index) so the topic layout matches production. The interval -I is per client in milliseconds, so 5,000 devices at -I 5000 send one message every 5 seconds each — 1,000 msg/s in total:

docker run --rm -it --ulimit nofile=65535:65535 emqx/emqtt-bench pub \
  -h broker.connectmq.com -p 1883 \
  -c 5000 -i 5 -I 5000 \
  -t 'testmqtt/fanin/%i/telemetry' -s 200 -q 1 \
  -u "$MQTT_USER" -P "$MQTT_PASS"

To increase the load, lower -I (5000 → 2000 → 1000 ms gives 2,500 and 5,000 msg/s) rather than adding connections: you are testing the consumer, not the connection count. Before you start, check your wildcard actually matches the device topics with the topic matcher — a filter typo looks exactly like a consumer that receives nothing.

Step 2: a consumer that shows its backlog

A production-style consumer should never do slow work inside on_message; that blocks the network loop, which stops reading the socket and pushes the backlog into the broker where you cannot see it. This paho-mqtt 2.x consumer hands each message to a queue, processes it in worker threads with a simulated per-message cost, and prints receive rate, processing rate and backlog every second:

import json, queue, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = "testmqtt/fanin/+/telemetry"
SHARED = "--shared" in sys.argv
TOPIC = f"$share/ingest/{FILTER}" if SHARED else FILTER
WORK_MS = 0.5        # simulated cost per message (parse + DB insert)
WORKERS = 4

inbox = queue.Queue(maxsize=100_000)
counts = {"rx": 0, "done": 0}
lags = []

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    counts["rx"] += 1
    inbox.put(msg.payload)       # never do slow work on the network thread

def worker():
    while True:
        payload = inbox.get()
        if payload[:1] == b"{":  # timestamped probe messages
            try:
                lags.append((time.time() - json.loads(payload)["t"]) * 1000)
            except (ValueError, KeyError):
                pass
        time.sleep(WORK_MS / 1000)
        counts["done"] += 1

for _ in range(WORKERS):
    threading.Thread(target=worker, daemon=True).start()

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                client_id=f"ingest-{uuid.uuid4().hex[:6]}", protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect(HOST, PORT)
c.loop_start()

while True:
    time.sleep(1)
    l = sorted(lags); lags.clear()
    p99 = f"{l[int(len(l) * 0.99) - 1]:.0f} ms" if l else "-"
    print(f"rx {counts['rx']:>6}/s  done {counts['done']:>6}/s  "
          f"backlog {inbox.qsize():>6}  probe lag p99 {p99}")
    counts.update(rx=0, done=0)

emqtt-bench payloads are random bytes, so they cannot carry a timestamp. To measure end-to-end lag, run a tiny probe on the same machine as the consumer (so both use the same wall clock) that publishes timestamped JSON into the same topic tree:

import json, time
import paho.mqtt.client as mqtt

p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
p.connect("broker.connectmq.com", 1883)
p.loop_start()
while True:
    p.publish("testmqtt/fanin/probe/telemetry", json.dumps({"t": time.time()}), qos=1)
    time.sleep(0.1)

Reading the numbers

Run the fleet at each step and watch the consumer output for a few minutes:

• rx ≈ done, backlog near zero, lag flat — the consumer keeps up. Move to the next step.
• rx ≈ ingress but done < rx, backlog growing — the network side is fine, processing is the bottleneck. Add workers, batch writes or speed up the sink.
• rx < ingress, backlog small — the single connection or the broker-side session is the limit (inflight window, per-client queue). This is where a shared subscription helps.
• Lag rising steadily — you are above the sustainable rate even if nothing is lost yet; the queue is just getting longer.
• Consumer disconnected under load — the broker hit its per-client limit or keep-alive expired because the network thread was blocked.

Step 3: scale out with shared subscriptions

When one consumer connection is the limit, start several consumers in the same shared subscription group. The broker then delivers each message to only one member of $share/ingest/...:

python ingest.py --shared   # terminal 1
python ingest.py --shared   # terminal 2 (and so on)

Repeat the step that failed before. What to verify:

1. The sum of done across consumers matches the fleet's ingress — no messages lost.
2. No duplicates — without $share, each consumer would receive every message and you would process everything twice. Count unique IDs if your payload carries them.
3. Even distribution — brokers use different strategies (round robin, random, sticky per client). Check that load is roughly balanced.
4. Failover — kill one consumer during the run; the others should absorb its share. Messages in flight to the killed consumer may be redelivered or lost depending on QoS and broker.
5. Ordering — per-device order is not guaranteed across the group. If it matters, design for it.

Shared subscriptions explains the semantics, and the shared subscription test covers distribution and failover tests in more depth.

Beyond one process

If the sink is the bottleneck and bursts are unavoidable — a fleet reconnecting after an outage replays a lot of buffered data at once — put a durable buffer between MQTT and storage. Bridging MQTT into Kafka is a common pattern: see MQTT to Kafka. Then test the burst case explicitly: stop the consumer for five minutes while the fleet runs, start it again and measure how long the backlog takes to drain. For more realistic device behaviour than emqtt-bench offers, use a device simulator with your real payloads.

Before a big run, a quick sanity check in the online MQTT client — subscribe to testmqtt/fanin/# and watch messages arrive — confirms that the fleet is publishing where you think it is.

Fan-in rarely breaks the broker; it breaks the consumer. Measure backlog and lag rather than raw receive rates, keep slow work off the network thread, and scale with shared subscriptions once one connection is not enough.

Frequently asked questions