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

Message Queue Protocols: A Complete Reference Guide

Introduction to Message Queue Protocols

Message queue protocols are the standardized rules and formats that govern how messages are structured, transmitted, routed, and consumed between producers and brokers, and between brokers and consumers. They define the wire-level communication contract that ensures interoperability across different languages, platforms, and implementations. Without these protocols, distributed systems would be forced into proprietary, vendor-locked communication patterns that are brittle and difficult to evolve.

Think of a message queue protocol the way you think of HTTP for web services: it is the agreed-upon language that clients and servers speak. Just as a REST API relies on HTTP verbs, status codes, and headers, a message queue relies on its protocol to define frames, exchanges, queues, acknowledgments, and delivery guarantees.

This reference guide covers the most widely adopted message queue protocols—AMQP, MQTT, STOMP, the Kafka wire protocol, and the JMS API—with practical code examples, comparison tables, and best practices drawn from real-world production systems.

What Is a Message Queue Protocol?

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

A message queue protocol is a specification that defines:

Protocols can be binary (AMQP, MQTT, Kafka) or text-based (STOMP). Binary protocols are generally more compact and performant, while text-based protocols are easier to debug and implement in simple clients. The choice of protocol shapes your system's latency profile, throughput ceiling, and operational complexity.

Why Message Queue Protocols Matter

Choosing the right protocol is not merely an academic exercise—it has direct consequences on production behavior:

In short, the protocol is the foundation on which your asynchronous architecture stands. A poor fit leads to costly workarounds, while a well-chosen protocol becomes an invisible enabler.

AMQP: Advanced Message Queuing Protocol

Overview

AMQP (Advanced Message Queuing Protocol) is a binary, wire-level protocol designed for enterprise messaging. The most widely deployed version is AMQP 0-9-1, which RabbitMQ has popularized. AMQP 1.0 is an OASIS and ISO standard that diverges significantly from 0-9-1 and is used by Apache Qpid and Azure Service Bus.

AMQP 0-9-1 defines a model of exchanges, bindings, and queues. Producers send messages to exchanges, which route them to queues based on binding rules. Consumers read from queues. This decoupling allows sophisticated routing patterns—direct, fanout, topic, and headers exchanges—without changing producer or consumer code.

Key Concepts

Code Example: AMQP with RabbitMQ in Node.js

// producer.js - Sends messages to a topic exchange
const amqp = require('amqplib');

async function produce() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare a topic exchange
  const exchange = 'order_events';
  await channel.assertExchange(exchange, 'topic', { durable: true });

  // Publish a message with a routing key
  const routingKey = 'order.created.europe';
  const message = Buffer.from(JSON.stringify({
    orderId: 'ORD-98765',
    customer: 'acme-corp',
    total: 1499.99,
    currency: 'EUR'
  }));

  channel.publish(exchange, routingKey, message, {
    persistent: true,
    contentType: 'application/json'
  });

  console.log(`Published to ${exchange} with key ${routingKey}`);

  // Graceful shutdown with publisher confirms
  setTimeout(() => {
    channel.close();
    connection.close();
  }, 500);
}

produce().catch(console.error);
// consumer.js - Consumes from a bound queue
const amqp = require('amqplib');

async function consume() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchange = 'order_events';
  await channel.assertExchange(exchange, 'topic', { durable: true });

  // Create a queue and bind with a pattern
  const queue = 'europe_orders_queue';
  await channel.assertQueue(queue, { durable: true });
  await channel.bindQueue(queue, exchange, 'order.created.europe');
  await channel.bindQueue(queue, exchange, 'order.updated.europe');

  // Set prefetch to limit unacknowledged messages
  channel.prefetch(10);

  console.log(`Waiting for messages on ${queue}`);

  channel.consume(queue, (msg) => {
    if (msg !== null) {
      const content = JSON.parse(msg.content.toString());
      console.log(`Received order ${content.orderId} for ${content.customer}`);

      // Process the message, then acknowledge
      // If processing fails, use channel.nack(msg, false, true) to requeue
      channel.ack(msg);
    }
  }, { noAck: false });
}

consume().catch(console.error);

Best Practices for AMQP

MQTT: Message Queuing Telemetry Transport

Overview

MQTT is a lightweight, binary publish-subscribe protocol designed for constrained devices and low-bandwidth, high-latency networks. It is the dominant protocol in IoT, automotive telematics, and mobile messaging applications. MQTT 3.1.1 is an OASIS standard, and MQTT 5.0 adds enhanced error reporting, session expiry, and message metadata.

Unlike AMQP, MQTT has no concept of exchanges or queues. Instead, it uses a flat topic namespace where the broker acts as a central dispatcher. Clients publish to topics and subscribe to topic filters (supporting wildcards: + for single-level and # for multi-level).

Key Concepts

Code Example: MQTT with Python (paho-mqtt)

# mqtt_publisher.py
import paho.mqtt.client as mqtt
import json
import time

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to broker")
    else:
        print(f"Connection failed with code {rc}")

def on_publish(client, userdata, mid):
    print(f"Message {mid} published")

client = mqtt.Client(client_id="sensor_gateway_01")
client.on_connect = on_connect
client.on_publish = on_publish

# Set Last Will: if this client disconnects abruptly,
# publish a status alert to the health topic
client.will_set(
    "sensors/gateway_01/status",
    payload="OFFLINE",
    qos=1,
    retain=True
)

# Connect with keep-alive of 60 seconds
client.connect("localhost", 1883, 60)

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

# Publish sensor readings
while True:
    payload = json.dumps({
        "temperature": 23.4,
        "humidity": 67.1,
        "timestamp": int(time.time())
    })
    
    # QoS 1 ensures the broker acknowledges receipt
    result = client.publish(
        "sensors/gateway_01/readings",
        payload=payload,
        qos=1,
        retain=False
    )
    print(f"Published with message id {result[1]}")
    time.sleep(10)
# mqtt_subscriber.py
import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected - subscribing to topics")
        # Subscribe to all sensor readings from any gateway
        client.subscribe("sensors/+/readings", qos=1)
        # Subscribe to status messages
        client.subscribe("sensors/+/status", qos=1)
    else:
        print(f"Connection failed: {rc}")

def on_message(client, userdata, msg):
    topic = msg.topic
    payload = msg.payload.decode("utf-8")
    
    # QoS level received
    qos = msg.qos
    retained = msg.retain
    
    print(f"Topic: {topic} | QoS: {qos} | Retained: {retained}")
    print(f"Payload: {payload}")
    print("-" * 50)

client = mqtt.Client(client_id="monitoring_dashboard")
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)
client.loop_forever()

Best Practices for MQTT

STOMP: Simple Text Oriented Messaging Protocol

Overview

STOMP (Simple Text Oriented Messaging Protocol) is a text-based, frame-oriented protocol that is intentionally simple and human-readable. It is often used as a lightweight alternative when AMQP feels too heavy or when clients are written in languages with limited binary protocol support. STOMP is supported by RabbitMQ (via plugin), ActiveMQ, and HornetQ.

Frames consist of a command (SEND, SUBSCRIBE, ACK, etc.), a set of headers, and an optional body separated by a blank line and terminated with a null byte. This simplicity makes STOMP easy to implement in shell scripts, curl-like tools, and simple HTTP-to-messaging bridges.

Code Example: STOMP over WebSocket in a Browser

<!-- Browser-based STOMP client using stomp.js -->
<script src="https://cdn.jsdelivr.net/npm/stompjs@2.3.3/lib/stomp.min.js"></script>
<script>
// Connect to RabbitMQ's STOMP-over-WebSocket plugin
const client = new Stomp.client('ws://localhost:15674/ws');

// Fallback for browsers without WebSocket
client.heartbeat.outgoing = 30000; // 30s client heartbeat
client.heartbeat.incoming = 0;     // No server heartbeat expected

const onConnect = function(frame) {
  console.log('Connected:', frame);
  
  // Subscribe to a destination
  const subscription = client.subscribe('/queue/notifications', function(message) {
    const body = JSON.parse(message.body);
    console.log('Received notification:', body);
    
    // Acknowledge the message (client-individual mode)
    message.ack();
  }, {
    ack: 'client-individual',
    'prefetch-count': 10
  });

  // Send a message
  client.send('/queue/notifications', {
    'content-type': 'application/json',
    'persistent': 'true'
  }, JSON.stringify({
    type: 'ALERT',
    severity: 'HIGH',
    message: 'Disk usage at 92%'
  }));

  console.log('Message sent');
};

const onError = function(error) {
  console.error('STOMP error:', error);
  // Reconnect after delay
  setTimeout(() => {
    client.connect('guest', 'guest', onConnect, onError, '/');
  }, 5000);
};

client.connect('guest', 'guest', onConnect, onError, '/');
</script>
# Python STOMP client using stomp.py
import stomp
import json
import time

class NotificationListener(stomp.ConnectionListener):
    def on_message(self, frame, headers, body):
        print(f"Headers: {headers}")
        print(f"Body: {body}")
        # Acknowledge the message
        ack_id = headers.get('ack')
        if ack_id:
            conn.ack(ack_id, headers['subscription'])
    
    def on_disconnected(self):
        print("Disconnected - will attempt reconnect")
        time.sleep(3)
        connect_and_subscribe()

def connect_and_subscribe():
    global conn
    conn = stomp.Connection([('localhost', 61613)])
    conn.set_listener('', NotificationListener())
    conn.connect('guest', 'guest', wait=True)
    
    # Subscribe with client-individual acknowledgment mode
    conn.subscribe(
        destination='/queue/notifications',
        id=1,
        ack='client-individual',
        headers={'prefetch-count': '5'}
    )
    print("Subscribed to /queue/notifications")

conn = stomp.Connection([('localhost', 61613)])
connect_and_subscribe()

# Keep the connection alive
while True:
    time.sleep(60)
    if not conn.is_connected():
        connect_and_subscribe()

Best Practices for STOMP

Apache Kafka Wire Protocol

Overview

Apache Kafka uses its own binary TCP protocol, designed from the ground up for high-throughput log-based messaging. Unlike AMQP or MQTT, Kafka's protocol is not an open standard governed by a standards body, but it is fully documented and has client implementations in dozens of languages.

The Kafka protocol is request-response based, with the client initiating all interactions. It uses a compact binary encoding with variable-length integers, tagged fields, and flexible versioning to support backward-compatible evolution. The protocol handles metadata discovery, partition leadership, consumer group coordination, and offset management natively.

Kafka's model is fundamentally different from traditional message queues: it is a distributed commit log where messages are appended to partitioned, immutable topic logs. Consumers use offsets to track their position, enabling replay and time-travel debugging.

Code Example: Kafka Producer and Consumer in Java

// KafkaProducerExample.java
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaProducerExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        
        // Enable idempotent producer for exactly-once semantics
        props.put("enable.idempotence", "true");
        props.put("acks", "all");
        props.put("retries", Integer.MAX_VALUE);
        props.put("max.in.flight.requests.per.connection", 5);
        
        Producer producer = new KafkaProducer<>(props);
        
        // Asynchronous send with callback
        ProducerRecord record = new ProducerRecord<>(
            "user_events",           // topic
            "user_123",              // key (ensures ordering for this user)
            "{\"action\":\"login\",\"timestamp\":\"2025-01-15T10:30:00Z\"}"
        );
        
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    System.out.printf(
                        "Sent to partition %d, offset %d%n",
                        metadata.partition(),
                        metadata.offset()
                    );
                } else {
                    System.err.println("Send failed: " + exception.getMessage());
                }
            }
        });
        
        // Synchronous send (blocks until acknowledged)
        RecordMetadata metadata = producer.send(record).get();
        System.out.printf("Sync send: partition %d, offset %d%n",
            metadata.partition(), metadata.offset());
        
        producer.flush();
        producer.close();
    }
}
// KafkaConsumerExample.java
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "user_event_processor");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        
        // Control offset commit strategy
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");
        props.put("max.poll.records", 50);
        
        Consumer consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("user_events"));
        
        try {
            while (true) {
                ConsumerRecords records = 
                    consumer.poll(Duration.ofMillis(1000));
                
                for (ConsumerRecord record : records) {
                    System.out.printf(
                        "Consumed: topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
                        record.topic(),
                        record.partition(),
                        record.offset(),
                        record.key(),
                        record.value()
                    );
                    
                    // Process the record...
                    processUserEvent(record.key(), record.value());
                }
                
                // Commit offsets after batch processing
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
    
    private static void processUserEvent(String key, String value) {
        // Your business logic here
        System.out.println("Processing event for user: " + key);
    }
}

Best Practices for Kafka Protocol

JMS: Java Message Service API

Overview

JMS (Java Message Service) is not a wire protocol but a Java API specification (JSR 914) that defines a standardized interface for messaging. It is included here because it is often discussed alongside wire protocols and can use various underlying protocols (AMQP, Kafka, or proprietary ones) via service provider implementations.

JMS defines two messaging domains: point-to-point (queues, with QueueConnectionFactory and Queue destinations) and publish-subscribe (topics, with TopicConnectionFactory and Topic destinations). It provides both synchronous and asynchronous message consumption, durable subscriptions, and message-driven beans for Jakarta EE containers.

Code Example: JMS with ActiveMQ

// JmsProducer.java
import jakarta.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsProducer {
    public static void main(String[] args) throws JMSException {
        // Create connection factory (ActiveMQ implements JMS)
        ConnectionFactory factory = new ActiveMQConnectionFactory(
            "tcp://localhost:61616"
        );
        
        Connection connection = factory.createConnection("admin", "admin");
        connection.start();
        
        // Create session (transacted=false, auto-acknowledge)
        Session session = connection.createSession(
            false,
            Session.AUTO_ACKNOWLEDGE
        );
        
        // Create destination and producer
        Destination destination = session.createQueue("ORDER_QUEUE");
        MessageProducer producer = session.createProducer(destination);
        
        // Set delivery mode for persistence
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        
        // Create a text message
        TextMessage message = session.createTextMessage(
            "{\"orderId\": \"ORD-12345\", \"amount\": 250.00}"
        );
        
        // Set JMS headers
        message.setJMSPriority(9);  // High priority
        message.setJMSExpiration(3600000); // 1 hour TTL
        
        // Add application properties for routing/filtering
        message.setStringProperty("orderType", "PRIORITY");
        message.setStringProperty("region", "US-EAST");
        
        producer.send(message);
        System.out.println("Message sent successfully");
        
        producer.close();
        session.close();
        connection.close();
    }
}
// JmsConsumer.java
import jakarta.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsConsumer {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory factory = new ActiveMQConnectionFactory(
            "tcp://localhost:61616"
        );
        
        Connection connection = factory.createConnection("admin", "admin");
        connection.start();
        
        // Client-acknowledge session for controlled acknowledgment
        Session session = connection.createSession(
            false,
            Session.CLIENT_ACKNOWLEDGE
        );
        
        Destination destination = session.createQueue("ORDER_QUEUE");
        MessageConsumer consumer = session.createConsumer(
            destination,
            "orderType = 'PRIORITY' AND region = 'US-EAST'"  // JMS selector
        );
        
        // Register a message listener for asynchronous consumption
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    if (message instanceof TextMessage) {
                        String body = ((TextMessage) message).getText();
                        System.out.println("Received: " + body);
                        
                        // Process the order...
                        processOrder(body);
                        
                        // Acknowledge after successful processing
                        message.acknowledge();
                    }
                } catch (JMSException e) {
                    System.err.println("Processing failed: " + e.getMessage());
                    // Do not acknowledge - message will be redelivered
                }
            }
        });
        
        // Keep the application running
        try {
            Thread.sleep(60000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        consumer.close();
        session.close();
        connection.close();
    }
    
    private static void processOrder(String orderJson) {
        System.out.println("Processing order: " + orderJson);
    }
}

Best Practices for JMS

Protocol Comparison Matrix

When selecting a protocol for your architecture, consider the following trade-offs across the major protocols:

AMQP 0-9-1 vs MQTT vs STOMP vs Kafka

How to Choose the Right Protocol

Follow this decision framework to narrow down your protocol choice:

General Best Practices Across All Protocols

Regardless of which protocol you adopt, these principles hold true in production:

Conclusion

Message queue protocols are the invisible infrastructure that shapes the reliability, performance, and evolvability of distributed systems. AMQP gives you enterprise-grade routing and transactional semantics. MQTT gives you lightweight, battery-friendly connectivity for the Internet of Things. STOMP gives you simplicity and browser-native real-time messaging. Kafka gives you an immutable, replayable event log that powers modern stream processing. JMS provides a mature Java API that abstracts away wire-level concerns.

The key insight is that no single protocol is universally superior. Each represents a set of design trade-offs optimized for specific use cases. The best architects understand these trade-offs deeply and match the protocol to the problem rather than forcing a familiar

🚀 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