← Back to DevBytes
Developer Tutorial July 11, 2026 16 min read

Implementing MQTT Protocol: From Theory to Practice

Understanding the MQTT Protocol

MQTT (Message Queuing Telemetry Transport) is a lightweight, publish-subscribe network protocol that transports messages between devices. It was invented by Andy Stanford-Clark of IBM and Arlen Nipper of Cirrus Link in 1999 to connect oil pipelines over unreliable satellite networks. Today, it has become the de facto standard protocol for Internet of Things (IoT) communication, powering everything from smart home devices to industrial sensor networks.

At its core, MQTT operates on a simple principle: a broker sits at the center, and clients connect to it to either publish messages or subscribe to topics they're interested in. The broker handles message routing, decoupling producers from consumers entirely. A publisher never needs to know who consumes its data, and a subscriber never needs to know who produced it. This decoupling is what makes MQTT so powerful for distributed systems.

The Publish-Subscribe Model Explained

Unlike HTTP's request-response model, MQTT uses a push-based publish-subscribe pattern. Here's how it works:

When a publisher sends a message to topic sensors/outdoor/humidity, the broker instantly forwards that message to every client that has subscribed to that exact topic or used a wildcard matching it. The publisher receives no acknowledgment from subscribers—only from the broker, depending on the Quality of Service level chosen.

Why MQTT Matters for Modern Development

MQTT fills a critical gap that HTTP and other protocols cannot efficiently address. Here are the key reasons developers choose MQTT:

Core Protocol Concepts Deep Dive

Quality of Service (QoS) Levels

MQTT defines three QoS levels that control the delivery contract between a publisher and the broker, and separately between the broker and each subscriber:

Retained Messages

A publisher can set the retain flag on any message. The broker stores the last retained message for each topic. When a new client subscribes, it immediately receives that stored message—even if no publisher has sent anything recently. This is perfect for publishing static configuration data or a sensor's last known state so new subscribers don't have to wait for the next update cycle.

Last Will and Testament (LWT)

Every client can specify a "last will" message during connection. If the client disconnects unexpectedly (without a clean DISCONNECT packet), the broker publishes this LWT message to a specified topic. This allows the system to detect offline devices and take action—for example, a client could publish device-123/status with payload "offline" as its LWT, and publish "online" as a retained message upon successful connection.

Persistent Sessions and Clean Start

In MQTT 5 (and MQTT 3.1.1's "clean session" flag), a client can request a persistent session. The broker stores the client's subscriptions, unacknowledged QoS 1 and QoS 2 messages, and queued messages while the client is offline. When the client reconnects, all pending messages are delivered. A clean start resets this state entirely.

Setting Up Your First MQTT Broker

Before writing client code, you need a broker. The most popular open-source broker is Eclipse Mosquitto. Here's how to install and run it on a Linux system:

# Install Mosquitto broker and client utilities
sudo apt update
sudo apt install mosquitto mosquitto-clients -y

# Start the broker as a service
sudo systemctl start mosquitto
sudo systemctl enable mosquitto

# Verify it's running
sudo systemctl status mosquitto

# Test with command-line clients (in two terminals):
# Terminal 1: Subscriber
mosquitto_sub -t "test/hello" -v

# Terminal 2: Publisher
mosquitto_pub -t "test/hello" -m "World's first MQTT message!"

For development on macOS, use Homebrew: brew install mosquitto. On Windows, download the installer from the Mosquitto website. For a quick cloud-based option without local installation, use the free public test broker at test.mosquitto.org (port 1883 for unencrypted, port 8883 for TLS) or broker.hivemq.com (port 1883).

Implementing an MQTT Client in Python

Python's paho-mqtt library is the most widely used MQTT client implementation. Let's build a complete publisher and subscriber from scratch.

First, install the library:

pip install paho-mqtt

Building a Robust MQTT Publisher

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

# --- Configuration ---
BROKER_ADDRESS = "localhost"
BROKER_PORT = 1883
TOPIC_BASE = "home/sensors"
CLIENT_ID = "python-publisher-001"

# --- Callbacks ---
def on_connect(client, userdata, flags, reason_code, properties):
    """Called when the client connects to the broker."""
    if reason_code == 0:
        print("✓ Connected successfully to broker")
        # Publish a retained "online" status message
        client.publish(
            f"{TOPIC_BASE}/publisher-status",
            payload="online",
            qos=1,
            retain=True
        )
    else:
        print(f"✗ Connection failed with reason code: {reason_code}")
        # Common reason codes:
        # 1: Connection refused - incorrect protocol version
        # 2: Connection refused - incorrect client identifier
        # 3: Connection refused - server unavailable
        # 4: Connection refused - bad username or password
        # 5: Connection refused - not authorized

def on_publish(client, userdata, mid, reason_code, properties):
    """Called when a published message is acknowledged by the broker (QoS > 0)."""
    print(f"  ↳ Message {mid} acknowledged (reason_code={reason_code})")

def on_disconnect(client, userdata, flags, reason_code, properties):
    """Called when the client disconnects."""
    if reason_code == 0:
        print("✓ Disconnected cleanly")
    else:
        print(f"✗ Unexpected disconnection! Reason code: {reason_code}")
        # Automatic reconnect is handled by the loop, but you can add logic here

# --- Create client with modern callback API (MQTT v5) ---
client = mqtt.Client(
    client_id=CLIENT_ID,
    protocol=mqtt.MQTTv5,
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)

# Assign callbacks
client.on_connect = on_connect
client.on_publish = on_publish
client.on_disconnect = on_disconnect

# --- Set Last Will and Testament ---
client.will_set(
    topic=f"{TOPIC_BASE}/publisher-status",
    payload="offline",
    qos=1,
    retain=True
)
print("LWT configured: will publish 'offline' if unexpectedly disconnected")

# --- Connect to broker ---
print(f"Connecting to {BROKER_ADDRESS}:{BROKER_PORT}...")
client.connect(BROKER_ADDRESS, BROKER_PORT, keepalive=60)

# Start the network loop in a background thread
client.loop_start()

# --- Publish loop ---
try:
    counter = 0
    while True:
        counter += 1
        
        # Create a realistic sensor payload
        sensor_data = {
            "timestamp": int(time.time()),
            "device_id": CLIENT_ID,
            "temperature_c": round(22.5 + random.uniform(-1.5, 1.5), 1),
            "humidity_pct": round(55 + random.uniform(-8, 8), 1),
            "battery_v": round(3.7 - counter * 0.001, 2),
            "sequence_number": counter
        }
        
        payload = json.dumps(sensor_data)
        
        # Publish with QoS 1 for reliable delivery
        result = client.publish(
            topic=f"{TOPIC_BASE}/environment",
            payload=payload,
            qos=1,
            retain=False
        )
        
        print(f"Published #{counter}: {payload[:80]}...")
        
        # Publish a high-priority alert every 10 messages (QoS 2 demo)
        if counter % 10 == 0:
            alert_payload = json.dumps({
                "type": "HEARTBEAT",
                "device": CLIENT_ID,
                "sequence": counter,
                "status": "nominal"
            })
            client.publish(
                f"{TOPIC_BASE}/alerts",
                payload=alert_payload,
                qos=2,
                retain=False
            )
            print(f"  ↳ Sent heartbeat alert with QoS 2")
        
        time.sleep(2)

except KeyboardInterrupt:
    print("\nShutdown requested...")
    
    # Publish offline status before disconnecting
    client.publish(
        f"{TOPIC_BASE}/publisher-status",
        payload="offline",
        qos=1,
        retain=True
    )
    print("Published final offline status")
    
    # Clean shutdown
    client.loop_stop()
    client.disconnect()
    print("Disconnected. Goodbye!")

Building a Resilient MQTT Subscriber

import paho.mqtt.client as mqtt
import json
import time
from collections import defaultdict

# --- Configuration ---
BROKER_ADDRESS = "localhost"
BROKER_PORT = 1883
TOPIC_FILTERS = [
    ("home/sensors/environment", 1),   # QoS 1 subscription
    ("home/sensors/alerts", 2),        # QoS 2 subscription
    ("home/sensors/publisher-status", 1)
]
CLIENT_ID = "python-subscriber-001"

# --- Statistics tracking ---
stats = defaultdict(lambda: {"count": 0, "last_payload": None, "last_timestamp": 0})

# --- Callbacks ---
def on_connect(client, userdata, flags, reason_code, properties):
    """Called on broker connection. Subscribe to topics here."""
    if reason_code != 0:
        print(f"✗ Connection failed: {reason_code}")
        return
    
    print("✓ Connected. Subscribing to topics...")
    
    for topic_filter, qos in TOPIC_FILTERS:
        # Subscribe to each topic
        result, mid = client.subscribe(topic_filter, qos)
        if result == mqtt.MQTT_ERR_SUCCESS:
            print(f"  ✓ Subscribed to '{topic_filter}' (QoS {qos})")
        else:
            print(f"  ✗ Failed to subscribe to '{topic_filter}'")

def on_message(client, userdata, message):
    """Called for every received message that matches subscriptions."""
    topic = message.topic
    payload_str = message.payload.decode("utf-8")
    qos = message.qos
    retained = message.retain
    
    # Update stats
    stats[topic]["count"] += 1
    stats[topic]["last_timestamp"] = time.time()
    
    # Try to parse JSON
    try:
        data = json.loads(payload_str)
        stats[topic]["last_payload"] = data
        is_json = True
    except json.JSONDecodeError:
        stats[topic]["last_payload"] = payload_str
        is_json = False
    
    # Pretty print
    print(f"\n📨 [{topic}] QoS={qos} Retained={retained}")
    if is_json:
        print(f"   Temperature: {data.get('temperature_c', 'N/A')}°C")
        print(f"   Humidity:    {data.get('humidity_pct', 'N/A')}%")
        if 'type' in data:
            print(f"   Alert:       {data.get('type')} - {data.get('status')}")
    else:
        print(f"   Raw: {payload_str}")
    
    print(f"   Total messages on this topic: {stats[topic]['count']}")

def on_subscribe(client, userdata, mid, reason_codes, properties):
    """Called when a subscription is acknowledged."""
    print(f"  ↳ Subscription {mid} acknowledged: {reason_codes}")

# --- Create client ---
client = mqtt.Client(
    client_id=CLIENT_ID,
    protocol=mqtt.MQTTv5,
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)

client.on_connect = on_connect
client.on_message = on_message
client.on_subscribe = on_subscribe

# --- Connect and loop forever ---
print(f"Connecting to {BROKER_ADDRESS}:{BROKER_PORT}...")
client.connect(BROKER_ADDRESS, BROKER_PORT, keepalive=60)

print("Entering listening loop. Press Ctrl+C to exit.\n")
client.loop_forever()

Understanding MQTT Topic Syntax and Wildcards

MQTT topics are hierarchical strings separated by forward slashes. They do not support regex, but offer two powerful wildcard mechanisms:

Here's a practical example demonstrating wildcard subscriptions:

import paho.mqtt.client as mqtt

def on_message(client, userdata, message):
    print(f"[{message.topic}] → {message.payload.decode()}")

client = mqtt.Client(
    client_id="wildcard-demo",
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)
client.on_message = on_message
client.connect("localhost", 1883)

# Subscribe to all temperature sensors in any room
client.subscribe("home/+/temperature", qos=1)

# Subscribe to ALL topics under 'home' (useful for debugging)
client.subscribe("home/#", qos=1)

print("Listening for all home/* topics...")
client.loop_forever()

# Example topics that would be received:
# home/living-room/temperature ✓
# home/kitchen/temperature ✓
# home/garage/humidity ✓ (via #)
# home/living-room/sensor/battery ✓ (via #)
# office/temperature ✗ (not under home)

Implementing an MQTT Client in Node.js

For server-side JavaScript applications, the mqtt.js library provides a clean asynchronous API. Install it with:

npm install mqtt

Here's a complete Node.js client that both publishes sensor data and subscribes to commands:

const mqtt = require('mqtt');

// --- Configuration ---
const BROKER_URL = 'mqtt://localhost:1883';
const CLIENT_ID = 'nodejs-device-001';
const TOPICS = {
    telemetry: 'devices/nodejs-001/telemetry',
    commands: 'devices/nodejs-001/commands',
    status: 'devices/nodejs-001/status'
};

// --- Connect with options ---
const client = mqtt.connect(BROKER_URL, {
    clientId: CLIENT_ID,
    clean: false,           // Persistent session
    keepalive: 30,          // Ping every 30 seconds
    will: {                 // Last Will and Testament
        topic: TOPICS.status,
        payload: JSON.stringify({ status: 'offline', reason: 'connection-lost' }),
        qos: 1,
        retain: true
    }
});

// --- Event handlers ---
client.on('connect', () => {
    console.log('✓ Connected to broker');
    
    // Publish online status as retained message
    client.publish(TOPICS.status, JSON.stringify({
        status: 'online',
        timestamp: Date.now(),
        clientId: CLIENT_ID
    }), { qos: 1, retain: true });
    
    // Subscribe to command topic
    client.subscribe(TOPICS.commands, { qos: 2 }, (err, granted) => {
        if (err) {
            console.error('✗ Subscription error:', err);
        } else {
            console.log(`✓ Subscribed to ${TOPICS.commands} (QoS ${granted[0].qos})`);
        }
    });
});

client.on('message', (topic, payload, packet) => {
    // packet contains qos, retain, dup flags
    const message = payload.toString();
    console.log(`\n📨 [${topic}] QoS=${packet.qos} Retain=${packet.retain}`);
    
    try {
        const data = JSON.parse(message);
        
        // Handle commands
        if (topic === TOPICS.commands) {
            console.log(`  Command received: ${data.command}`);
            handleCommand(data);
        }
    } catch (e) {
        console.log(`  Raw message: ${message}`);
    }
});

client.on('reconnect', () => {
    console.log('⟳ Attempting to reconnect...');
});

client.on('close', () => {
    console.log('✗ Connection closed');
});

client.on('error', (err) => {
    console.error('✗ Client error:', err.message);
});

// --- Command handler ---
function handleCommand(cmd) {
    switch (cmd.command) {
        case 'reboot':
            console.log('  → Rebooting device...');
            // Send acknowledgment
            client.publish(`${TOPICS.commands}/response`, JSON.stringify({
                command: cmd.command,
                result: 'executing',
                timestamp: Date.now()
            }), { qos: 1 });
            break;
        case 'update_config':
            console.log(`  → Applying config: ${JSON.stringify(cmd.params)}`);
            client.publish(`${TOPICS.commands}/response`, JSON.stringify({
                command: cmd.command,
                result: 'success',
                timestamp: Date.now()
            }), { qos: 1 });
            break;
        default:
            console.log(`  → Unknown command: ${cmd.command}`);
    }
}

// --- Periodic telemetry publishing ---
let counter = 0;
const telemetryInterval = setInterval(() => {
    counter++;
    
    const telemetry = {
        timestamp: Date.now(),
        sequence: counter,
        cpu_temp: Math.round(45 + Math.random() * 15),
        memory_used_mb: Math.round(120 + Math.random() * 80),
        uptime_seconds: process.uptime()
    };
    
    client.publish(TOPICS.telemetry, JSON.stringify(telemetry), { qos: 1 });
    console.log(`↗ Published telemetry #${counter}`);
}, 5000);

// --- Graceful shutdown ---
process.on('SIGINT', () => {
    console.log('\nShutting down gracefully...');
    clearInterval(telemetryInterval);
    
    // Publish offline status
    client.publish(TOPICS.status, JSON.stringify({
        status: 'offline',
        reason: 'clean-shutdown',
        timestamp: Date.now()
    }), { qos: 1, retain: true }, () => {
        client.end(true, () => {
            console.log('✓ Disconnected cleanly');
            process.exit(0);
        });
    });
});

Securing Your MQTT Deployment

A production MQTT setup must implement proper security. Here's how to configure Mosquitto with TLS and authentication:

# Generate a CA certificate and server key (simplified for development)
# For production, use a proper CA like Let's Encrypt or your organization's PKI

# 1. Generate CA private key
openssl genrsa -out ca.key 2048

# 2. Create self-signed CA certificate
openssl req -new -x509 -days 3650 -key ca.key -out ca.crt \
  -subj "/C=US/ST=State/L=City/O=Org/CN=MQTT-CA"

# 3. Generate server private key
openssl genrsa -out server.key 2048

# 4. Create server certificate signing request
openssl req -new -key server.key -out server.csr \
  -subj "/C=US/ST=State/L=City/O=Org/CN=mqtt.example.com"

# 5. Sign the server certificate with CA
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key \
  -CAcreateserial -out server.crt -days 365

# 6. Copy certificates to Mosquitto directory
sudo cp ca.crt server.crt server.key /etc/mosquitto/certs/
sudo chmod 600 /etc/mosquitto/certs/server.key

# 7. Create password file for authentication
sudo mosquitto_passwd -c /etc/mosquitto/passwd admin
# Enter password when prompted

# 8. Configure Mosquitto (/etc/mosquitto/conf.d/secure.conf)
cat << 'EOF' | sudo tee /etc/mosquitto/conf.d/secure.conf
# Listen on standard MQTT port with TLS
listener 8883
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
cafile /etc/mosquitto/certs/ca.crt

# Require client certificates (mutual TLS - most secure)
# use_identity_as_username true
# require_certificate true

# Password authentication
password_file /etc/mosquitto/passwd
allow_anonymous false

# ACL: restrict topic access
acl_file /etc/mosquitto/acl.conf
EOF

# 9. Create ACL file
cat << 'EOF' | sudo tee /etc/mosquitto/acl.conf
# Admin can read/write everything
user admin
topic readwrite #

# Sensors can only publish to their own topics
user sensor-livingroom
topic write home/living-room/#
topic read home/living-room/#

# Dashboard user can only read
user dashboard
topic read home/#
EOF

# 10. Restart Mosquitto
sudo systemctl restart mosquitto

And here's the corresponding Python client configured for TLS with authentication:

import paho.mqtt.client as mqtt
import ssl

# --- Secure client configuration ---
client = mqtt.Client(
    client_id="secure-publisher",
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)

# Set username and password
client.username_pw_set("sensor-livingroom", "your-password-here")

# Configure TLS
client.tls_set(
    ca_certs="/etc/mosquitto/certs/ca.crt",      # Verify broker's certificate
    certfile=None,                                # Client cert (if using mutual TLS)
    keyfile=None,                                 # Client key (if using mutual TLS)
    cert_reqs=ssl.CERT_REQUIRED,                  # Require broker certificate verification
    tls_version=ssl.PROTOCOL_TLSv1_2,
    ciphers=None
)

# Optional: skip hostname verification (only for testing with self-signed certs)
# client.tls_insecure_set(False)  # Keep True only if CN doesn't match

client.on_connect = lambda c, u, f, rc, p: print(
    "✓ Secure connection established" if rc == 0 else f"✗ Failed: {rc}"
)

# Connect using TLS port
client.connect("mqtt.example.com", port=8883, keepalive=60)
client.loop_start()

# Publish securely
client.publish("home/living-room/temperature", payload="23.5", qos=1, retain=False)

# Keep alive for demonstration
import time
time.sleep(5)
client.loop_stop()
client.disconnect()

MQTT 5 vs MQTT 3.1.1 — What to Use and Why

MQTT 5, released in 2019, adds significant improvements over the widely deployed MQTT 3.1.1. Here's a comparison to guide your choice:

When starting a new project today, use MQTT 5 unless you need compatibility with older third-party libraries that only support 3.1.1. All major brokers (Mosquitto 2.0+, HiveMQ, EMQX, VerneMQ) and client libraries support MQTT 5.

Best Practices for Production MQTT Systems

1. Topic Naming Conventions

Design your topic hierarchy with intention. A well-structured topic tree makes wildcard subscriptions useful and ACL rules manageable:

# Good topic structure
<organization>/<site>/<area>/<device-type>/<device-id>/<metric>

# Concrete example
acme-corp/warehouse-3/zone-b/forklift/fork-12/battery-level
acme-corp/warehouse-3/zone-b/forklift/fork-12/location

# Avoid these patterns:
# - Leading slash: /acme-corp/warehouse  (creates empty first level)
# - Trailing slash: acme-corp/warehouse/
# - Special characters: acme-corp/warehouse#3/zone-b  (# has special meaning)
# - Extremely short, ambiguous names: a/b/c

2. Choose QoS Levels Wisely

3. Handle Reconnection and State

Always implement proper reconnection logic. In most client libraries, the loop handles reconnection automatically, but you should still design for it:

def create_resilient_client():
    client = mqtt.Client(
        client_id="resilient-client",
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        clean_start=False  # Keep session state across reconnects
    )
    
    # Set session expiry to 2 hours (MQTT 5 only)
    client.connect_properties = {
        'session_expiry_interval': 7200  # seconds
    }
    
    def on_disconnect(client, userdata, flags, rc, props):
        if rc != 0:
            print(f"Unexpected disconnect (rc={rc}). Auto-reconnect is active.")
            # The loop will try to reconnect automatically
            # Queue any critical messages here if needed
    
    client.on_disconnect = on_disconnect
    
    # Use reconnect_delay_set for custom backoff (some libs support this)
    # client.reconnect_delay_set(min_delay=1, max_delay=120)
    
    return client

4. Monitor Broker Health

Use the $SYS topic hierarchy (available in Mosquitto) to monitor broker metrics:

# Subscribe to broker statistics
mosquitto_sub -t '$SYS/#' -v

# Key metrics to watch:
# $SYS/broker/clients/connected        - Active client count
# $SYS/broker/messages/received        - Messages received per second
# $SYS/broker/messages/sent            - Messages sent per second  
# $SYS/broker/load/messages/received/15min  - 15-minute message load average
# $SYS/broker/memory/current           - Current memory usage

5. Keep Payloads Small

MQTT excels with small payloads. A good rule of thumb is to keep messages under 1KB. If you need to transmit large data, consider:

6. Test with Realistic Network Conditions

Don't only test on localhost. Simulate latency, packet loss, and disconnections:

# Use Linux netem to simulate poor network conditions
# Add 200ms latency and 5% packet loss
sudo tc qdisc add dev eth0 root netem delay 200ms loss 5%

# Remove the simulation
sudo tc qdisc del dev eth0 root

# Or use a remote broker with real internet latency
# test.mosquitto.org is excellent for this

7. Use Retained Messages Strategically

Retained messages are perfect for state topics but can cause confusion if overused. A good pattern:

To clear a retained message, publish an empty payload with the retain flag set to that topic:

# Clear a retained message
mosquitto_pub -t 'device/sensor-01/config

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles