Hey there! Calculating MQTT queue latency (the time a message spends waiting in the broker's queue before being consumed) is totally straightforward once you nail down tracking two key timestamps: when the message arrives at the broker, and when it's picked up by the consumer. Let's walk through how to make this happen.

Core Concept #

Queue latency = Consumer ingestion timestamp - Message arrival timestamp at the broker

The trick is embedding the arrival timestamp directly with the message (or via MQTT metadata) so the consumer can pull it out and compute the difference.

Step 1: Add Arrival Timestamps on the Producer Side #

When your producer sends a message to the MQTT broker, attach a precise UTC timestamp right then—this marks when the message enters the queue pipeline. You have two clean ways to do this:

Option A: Embed in the Message Payload #

Keep it simple by adding the timestamp to your message's data structure (great for MQTT 3.1/3.1.1 compatibility):

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

def on_connect(client, userdata, flags, rc):
    print(f"Connected to broker: {rc}")

client = mqtt.Client()
client.on_connect = on_connect
client.connect("your-broker-address", 1883, 60)

# Generate millisecond-precise UTC timestamp
arrival_time_ms = time.time() * 1000
# Package timestamp with your business data
message = {
    "sensor_reading": 24.5,
    "arrival_time_ms": arrival_time_ms
}

# Publish as JSON string
client.publish("sensors/temperature", json.dumps(message))
client.loop_forever()

Option B: Use MQTT 5.0 User Properties #

If you're on MQTT 5.0, use user properties to keep your payload clean (no mixing metadata with business data):

import paho.mqtt.client as mqtt
import time

client = mqtt.Client(protocol=mqtt.MQTTv5)
client.connect("your-broker-address", 1883, 60)

# Attach timestamp as a user property
client.publish(
    "sensors/temperature",
    payload="24.5",
    properties=mqtt.Properties(
        mqtt.PROPERTY_USER_PROPERTY,
        [("arrival_time_ms", str(time.time() * 1000))]
    )
)
client.loop_forever()

Step 2: Calculate Latency on the Consumer Side #

When the consumer receives the message, record the current time (ingestion timestamp), pull out the arrival timestamp, and subtract to get latency:

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

def on_message(client, userdata, msg):
    # Record when the consumer picks up the message
    ingestion_time_ms = time.time() * 1000

    # Option 1: Parse timestamp from payload
    payload = json.loads(msg.payload.decode())
    arrival_time_ms = payload.get("arrival_time_ms")

    # Option 2: Pull timestamp from MQTT 5.0 user properties
    # arrival_time_ms = float(msg.properties.user_property[0][1])

    if arrival_time_ms:
        latency_ms = ingestion_time_ms - arrival_time_ms
        print(f"Queue latency: {latency_ms:.2f} ms")
    else:
        print("Warning: No arrival timestamp found in message")

client = mqtt.Client()
client.on_message = on_message
client.connect("your-broker-address", 1883, 60)
client.subscribe("sensors/temperature")
client.loop_forever()

Critical Notes to Avoid Headaches #

• Clock Sync is Non-Negotiable: If your producer and consumer clocks are out of sync, your latency calculations will be garbage. Use NTP to keep all nodes' UTC clocks aligned.
• Stick to High Precision: Use millisecond (or microsecond) timestamps—second-level precision won't cut it for low-latency use cases.
• Broker-Side Timestamps (Advanced): If you want even more accuracy (e.g., account for producer-to-broker network delay), you can configure your broker to add a timestamp when it receives the message. This requires broker-specific plugins (like EMQ X Rules Engine or Mosquitto custom extensions) but eliminates producer clock dependency.
• Error Handling: Always add checks for missing or invalid timestamps to avoid crashes in production.

Bonus: Monitor Latency at Scale #

For high-throughput systems, store calculated latency values in a time-series database (like InfluxDB) and use tools like Grafana to build dashboards for tracking average latency, peak delays, and trends over time.

内容的提问来源于stack exchange，提问作者Amarjit Dhillon