RabbitMQ #

Not every messaging need requires a platform as complex as Kafka. When what you need is flexible message routing, a reliable task queue, or request-reply communication between services, RabbitMQ is a far more natural choice. RabbitMQ implements the AMQP (Advanced Message Queuing Protocol) and uses a push model — the broker actively pushes messages to consumers, unlike Kafka’s pull model. The result is lower latency for scenarios where each message needs to be processed immediately and then deleted. In the Java ecosystem, RabbitMQ’s official AMQP Client library is complete and easy to integrate.

RabbitMQ Architecture #

RabbitMQ’s working model is very different from Kafka. It’s important to understand this difference before writing code, because the model determines how you design your system.

Core Components #

RabbitMQ has four core components that interact with each other:

flowchart LR
    P[Producer] --> EX[Exchange]
    EX -->|binding key| Q1[Queue A]
    EX -->|binding key| Q2[Queue B]
    Q1 --> C1[Consumer 1]
    Q2 --> C2[Consumer 2]
    Q2 --> C3[Consumer 3]

Producer sends messages to an exchange, not directly to a queue. The producer doesn’t need to know which queue will receive its messages.

Exchange is the router. It receives messages from producers and decides which queue to forward each message to, based on the exchange type and binding key.

Queue is where messages wait until a consumer picks them up. Unlike Kafka, messages that have been consumed and acknowledged are deleted from the queue.

Binding is the rule that connects an exchange to a queue. A binding can include a routing key that the exchange uses to filter messages.

Virtual Hosts #

RabbitMQ supports virtual hosts (vhosts) — namespaces that separate exchanges, queues, and permissions. This allows a single RabbitMQ instance to serve several isolated applications, similar to databases within a single database server.

RabbitMQ Instance
├── vhost: /              ← default vhost
├── vhost: /ecommerce
│   ├── exchange: order-exchange
│   └── queue: order-queue
└── vhost: /payments
    ├── exchange: payment-exchange
    └── queue: payment-queue

Setting Up Dependencies #

Add the AMQP client to your Java project. For Maven:

<dependencies>
    <!-- RabbitMQ Java AMQP Client -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.21.0</version>
    </dependency>

    <!-- SLF4J for logging -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>2.0.12</version>
    </dependency>
</dependencies>

For Gradle:

dependencies {
    implementation 'com.rabbitmq:amqp-client:5.21.0'
    implementation 'org.slf4j:slf4j-simple:2.0.12'
}

To run RabbitMQ locally, Docker is the fastest way:

# Run RabbitMQ with the Management UI on port 15672
docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin \
  rabbitmq:3.13-management

# Access the Management UI at: http://localhost:15672
# Username: admin, Password: admin

Connections and Channels #

Before sending or receiving messages, you need to create a connection to RabbitMQ. RabbitMQ uses the connection and channel concepts, which are important to understand.

Connection is a TCP connection to the broker. Creating a connection is expensive — it involves a TCP handshake and protocol negotiation. In an application, you usually have one connection per process or per thread pool.

Channel is a “virtual connection” within a connection. A channel is a lightweight unit of work. All AMQP operations (publish, consume, declare) go through channels. Use one channel per thread to avoid concurrency issues.

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Channel;

public class RabbitMQConnection {

    public static ConnectionFactory createFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");

        // Automatic connection recovery if the connection drops
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000); // try reconnecting every 5 seconds

        // Heartbeat — detect dead connections faster
        factory.setRequestedHeartbeat(30); // 30 seconds

        return factory;
    }

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = createFactory();

        // try-with-resources to close the connection automatically
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            System.out.println("Connected to RabbitMQ!");
            System.out.println("Channel number: " + channel.getChannelNumber());
        }
    }
}

Don’t create a new connection for every message you send. This is very inefficient and can make RabbitMQ run out of file descriptors. Create one connection and use different channels for each thread.

Exchange Types #

Choosing the exchange type is the most important design decision in RabbitMQ. Each type has different routing semantics.

Direct Exchange #

A direct exchange delivers messages to queues whose binding key exactly matches the message’s routing key. This is the simplest type — suitable for task queues and routing by task type.

flowchart LR
    P[Producer] -->|routing_key=email| DE["Direct Exchange\norder-direct"]
    DE -->|binding: email| QE[Queue: email-queue]
    DE -->|binding: sms| QS[Queue: sms-queue]
    QE --> C1[Email Service]
    QS --> C2[SMS Service]

public class DirectExchangeDemo {

    private static final String EXCHANGE_NAME = "notification-direct";
    private static final String EMAIL_QUEUE = "email-queue";
    private static final String SMS_QUEUE = "sms-queue";

    public static void setup(Channel channel) throws Exception {
        // Declare the exchange — idempotent, safe to call repeatedly
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true); // durable=true

        // Declare the queue — durable so it survives broker restarts
        channel.queueDeclare(EMAIL_QUEUE, true, false, false, null);
        channel.queueDeclare(SMS_QUEUE, true, false, false, null);

        // Binding: connect the queue to the exchange with a routing key
        channel.queueBind(EMAIL_QUEUE, EXCHANGE_NAME, "email");
        channel.queueBind(SMS_QUEUE, EXCHANGE_NAME, "sms");
    }

    public static void publish(Channel channel, String type, String message) throws Exception {
        // The message will only be sent to queues whose binding key is "email" or "sms"
        channel.basicPublish(
            EXCHANGE_NAME,
            type,           // routing key
            null,           // properties
            message.getBytes()
        );
        System.out.println("Sent to routing key '" + type + "': " + message);
    }
}

Fanout Exchange #

A fanout exchange ignores the routing key and delivers messages to all queues bound to it. Suitable for broadcasting — one event that many services need to know about.

flowchart LR
    P[Producer] -->|routing key ignored| FE["Fanout Exchange\norder-fanout"]
    FE --> Q1[Queue: analytics-queue]
    FE --> Q2[Queue: notification-queue]
    FE --> Q3[Queue: audit-queue]
    Q1 --> C1[Analytics Service]
    Q2 --> C2[Notification Service]
    Q3 --> C3[Audit Service]

public class FanoutExchangeDemo {

    private static final String EXCHANGE_NAME = "order-events-fanout";

    public static void setup(Channel channel) throws Exception {
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);

        // Each consumer creates its own queue with a unique name
        // exclusive=true — the queue is deleted when the consumer disconnects
        String analyticsQueue = channel.queueDeclare().getQueue();
        String notifQueue = channel.queueDeclare().getQueue();
        String auditQueue = channel.queueDeclare().getQueue();

        // Binding to a fanout exchange — the routing key is ignored (can be empty)
        channel.queueBind(analyticsQueue, EXCHANGE_NAME, "");
        channel.queueBind(notifQueue, EXCHANGE_NAME, "");
        channel.queueBind(auditQueue, EXCHANGE_NAME, "");
    }

    public static void publish(Channel channel, String orderJson) throws Exception {
        // Empty routing key — a fanout exchange ignores it
        channel.basicPublish(EXCHANGE_NAME, "", null, orderJson.getBytes());
    }
}

Topic Exchange #

A topic exchange matches routing keys against binding patterns using wildcards. It’s the most flexible type and the most commonly used in production systems.

Wildcard rules:

* — replaces exactly one word
# — replaces zero or more words
Words are separated by dots (.)

flowchart LR
    P[Producer] --> TE["Topic Exchange\napp-events"]
    TE -->|order.*| Q1[Queue: order-all]
    TE -->|order.created| Q2[Queue: order-created-only]
    TE -->|#.error| Q3[Queue: all-errors]

    P -->|order.created| TE
    P -->|order.shipped| TE
    P -->|payment.error| TE

public class TopicExchangeDemo {

    private static final String EXCHANGE_NAME = "app-events";

    public static void setup(Channel channel) throws Exception {
        channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);

        // Queue 1: all order events (order.created, order.shipped, order.cancelled, ...)
        channel.queueDeclare("order-all-queue", true, false, false, null);
        channel.queueBind("order-all-queue", EXCHANGE_NAME, "order.*");

        // Queue 2: only order created
        channel.queueDeclare("order-created-queue", true, false, false, null);
        channel.queueBind("order-created-queue", EXCHANGE_NAME, "order.created");

        // Queue 3: all errors from all services (payment.error, order.error, ...)
        channel.queueDeclare("all-errors-queue", true, false, false, null);
        channel.queueBind("all-errors-queue", EXCHANGE_NAME, "#.error");
    }

    public static void publish(Channel channel, String routingKey, String message) throws Exception {
        channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
        System.out.println("Published with routing key: " + routingKey);
    }

    // Usage examples:
    // publish(channel, "order.created", "...")   → goes to order-all-queue AND order-created-queue
    // publish(channel, "order.shipped", "...")   → goes to order-all-queue only
    // publish(channel, "payment.error", "...")   → goes to all-errors-queue only
    // publish(channel, "order.error", "...")     → goes to order-all-queue AND all-errors-queue
}

Exchange Type Comparison #

	Direct	Fanout	Topic	Headers
Routing based on	Exact match	Broadcast to all	Wildcard patterns	Message headers
Flexibility	Low	No routing	High	Very high
Use case	Task queues, per-type notifications	Event broadcasting	Events with hierarchy	Complex routing without naming
Performance	Fast	Fastest	Medium	Slowest

Producers with Message Properties #

RabbitMQ messages can include properties — metadata sent alongside the payload without having to be embedded in the message body.

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.MessageProperties;

public class ProducerWithProperties {

    public static void publishWithProperties(Channel channel, String queue, String body) throws Exception {
        // ✗ ANTI-PATTERN: non-persistent message — lost if the broker restarts
        channel.basicPublish("", queue, null, body.getBytes());

        // ✓ CORRECT: persistent message with MessageProperties.PERSISTENT_TEXT_PLAIN
        channel.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, body.getBytes());

        // For more complete properties, use the builder
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .deliveryMode(2)                    // 2 = persistent
            .priority(5)                        // 0-9, requires a queue with x-max-priority
            .correlationId("req-123")           // for the request-reply pattern
            .replyTo("reply-queue")             // for the request-reply pattern
            .expiration("60000")                // message TTL: 60 seconds (ms as a string)
            .messageId(java.util.UUID.randomUUID().toString())
            .timestamp(new java.util.Date())
            .headers(java.util.Map.of(
                "source-service", "order-service",
                "retry-count", 0
            ))
            .build();

        channel.basicPublish("order-exchange", "order.created", properties, body.getBytes());
    }
}

deliveryMode=2 (persistent) only truly guarantees durability if the queue is also declared as durable=true. Both must be active together. If the queue isn’t durable, persistent messages will still be lost when the broker restarts.

Consumers and Acknowledgments #

Consumers in RabbitMQ are push-based — the broker actively pushes messages to consumers. Acknowledgment (ack) is the mechanism that tells the broker a message was successfully processed.

Manual Acknowledgment #

import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.CancelCallback;

public class ManualAckConsumer {

    public static void startConsuming(Channel channel, String queueName) throws Exception {

        // ✗ ANTI-PATTERN: auto ack — the message is considered done as soon as it's delivered
        // If the consumer crashes after receiving but before finishing processing, the message is lost
        boolean autoAck = false; // always false for production

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            long deliveryTag = delivery.getEnvelope().getDeliveryTag();

            try {
                System.out.println("Processing: " + message);
                processMessage(message);

                // ✓ Ack after successful processing — the broker removes the message from the queue
                // multiple=false → only ack this message, not all previous ones
                channel.basicAck(deliveryTag, false);

            } catch (Exception e) {
                System.err.println("Failed to process: " + e.getMessage());

                // Nack — tell the broker the message failed to process
                // requeue=true  → return to the queue (careful: can cause an infinite loop!)
                // requeue=false → discard the message (or send to a DLX if configured)
                boolean requeue = isTransientError(e); // only requeue for transient errors
                channel.basicNack(deliveryTag, false, requeue);
            }
        };

        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("Consumer cancelled: " + consumerTag);
        };

        channel.basicConsume(queueName, autoAck, deliverCallback, cancelCallback);
    }

    private static void processMessage(String message) throws Exception {
        // business logic
    }

    private static boolean isTransientError(Exception e) {
        // Example: network errors are transient, JSON parsing errors aren't
        return e instanceof java.io.IOException;
    }
}

Prefetch — Controlling Consumer Load #

The prefetch count determines how many messages the broker may send to a consumer before the consumer sends an ack. This is a critical setting for fair load distribution.

// ✗ ANTI-PATTERN: without prefetch (default) — the broker sends all messages to the first consumer
// Fast and slow consumers get unbalanced loads
// channel.basicQos(0); // 0 = no limit

// ✓ CORRECT: prefetch=1 — the broker only sends 1 new message after the consumer acks the previous one
// Load is distributed based on each consumer's speed
channel.basicQos(1);

// For higher throughput with multiple consumers, raise the prefetch
// but consider memory usage
channel.basicQos(10); // send max 10 messages before needing an ack

sequenceDiagram
    participant Broker
    participant C1 as Consumer 1 (slow)
    participant C2 as Consumer 2 (fast)

    Note over Broker,C2: WITHOUT prefetch — unfair
    Broker->>C1: msg 1, 2, 3, 4, 5 (all at once)
    Broker->>C2: msg 6, 7, 8, 9, 10

    Note over Broker,C2: WITH prefetch=1 — fair
    Broker->>C1: msg 1
    Broker->>C2: msg 2
    C2-->>Broker: ack msg 2
    Broker->>C2: msg 3
    C2-->>Broker: ack msg 3
    Broker->>C2: msg 4
    C1-->>Broker: ack msg 1
    Broker->>C1: msg 5

Dead Letter Exchange (DLX) #

The Dead Letter Exchange is RabbitMQ’s official mechanism for handling messages that fail to process. A message becomes a “dead letter” when:

It’s nacked with requeue=false
Its message TTL expires
The queue is full (if x-max-length is set)

public class DLXSetup {

    public static void setupWithDLX(Channel channel) throws Exception {
        // 1. Create the Dead Letter Exchange
        String dlxExchange = "dlx.order-events";
        String dlxQueue = "dlq.order-events";

        channel.exchangeDeclare(dlxExchange, "direct", true);
        channel.queueDeclare(dlxQueue, true, false, false, null);
        channel.queueBind(dlxQueue, dlxExchange, "order-events"); // routing key must match

        // 2. Create the main queue with a reference to the DLX
        java.util.Map<String, Object> args = new java.util.HashMap<>();
        args.put("x-dead-letter-exchange", dlxExchange);      // route dead messages to the DLX
        args.put("x-dead-letter-routing-key", "order-events"); // routing key in the DLX
        args.put("x-message-ttl", 300_000);                   // 5-minute TTL (optional)

        channel.queueDeclare("order-events-queue", true, false, false, args);
        channel.queueBind("order-events-queue", "order-exchange", "order.*");
    }
}

flowchart TD
    P[Producer] --> EX[order-exchange]
    EX --> Q["order-events-queue\nx-dead-letter-exchange: dlx"]
    Q --> C{"Consumer\nprocessed successfully?"}
    C -- Yes\nbasicAck --> DONE[Message deleted]
    C -- No\nbasicNack requeue=false --> DLX[dlx.order-events]
    DLX --> DLQ[dlq.order-events]
    DLQ --> OPS["Ops Team / Monitoring"]

A consumer for the DLQ is usually run separately for investigation and manual reprocessing:

public class DLQConsumer {

    public static void monitorDLQ(Channel channel) throws Exception {
        channel.basicQos(1);

        channel.basicConsume("dlq.order-events", false,
            (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                long deliveryTag = delivery.getEnvelope().getDeliveryTag();

                // Log failure details
                var headers = delivery.getProperties().getHeaders();
                System.out.printf(
                    "[DLQ] Failed message — reason: %s, message: %s%n",
                    headers != null ? headers.get("x-death") : "unknown",
                    message
                );

                // Send an alert to the monitoring system (Slack, PagerDuty, etc.)
                sendAlert(message, headers);

                // Ack in the DLQ so it isn't reprocessed unintentionally
                channel.basicAck(deliveryTag, false);
            },
            consumerTag -> {}
        );
    }

    private static void sendAlert(String message, java.util.Map<String, Object> headers) {
        // notification implementation to the ops team
        System.out.println("ALERT: message entered the DLQ, manual investigation needed.");
    }
}

The Request-Reply Pattern (RPC over RabbitMQ) #

RabbitMQ is very well suited to implementing the request-reply pattern — a caller sends a request and waits for a response. This is what distinguishes RabbitMQ from Kafka: Kafka isn’t designed for this pattern.

sequenceDiagram
    participant Client
    participant RabbitMQ
    participant Server

    Client->>RabbitMQ: publish to "rpc-queue"\ncorrelationId=uuid-123\nreplyTo="amq.rabbitmq.reply-to"
    RabbitMQ->>Server: deliver request
    Server->>Server: process request
    Server->>RabbitMQ: publish response to replyTo\ncorrelationId=uuid-123
    RabbitMQ->>Client: deliver response
    Client->>Client: match correlationId → done

import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class RPCClient {

    private final Channel channel;

    public RPCClient(Channel channel) throws Exception {
        this.channel = channel;
    }

    public String call(String request, int timeoutSeconds) throws Exception {
        String correlationId = UUID.randomUUID().toString();

        // BlockingQueue to wait for the response from the consumer callback (a different thread)
        BlockingQueue<String> responseQueue = new ArrayBlockingQueue<>(1);

        // "amq.rabbitmq.reply-to" is RabbitMQ's built-in pseudo-queue for Direct Reply-to
        // More efficient than creating a new temporary queue per request
        String replyTo = "amq.rabbitmq.reply-to";

        // Subscribe to the reply queue before publishing the request
        channel.basicConsume(replyTo, true, // auto ack for the reply queue
            (consumerTag, delivery) -> {
                if (correlationId.equals(delivery.getProperties().getCorrelationId())) {
                    responseQueue.offer(new String(delivery.getBody(), "UTF-8"));
                }
            },
            consumerTag -> {}
        );

        // Publish the request
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .correlationId(correlationId)
            .replyTo(replyTo)
            .build();

        channel.basicPublish("", "rpc-queue", props, request.getBytes());

        // Wait for the response with a timeout
        String response = responseQueue.poll(timeoutSeconds, TimeUnit.SECONDS);
        if (response == null) {
            throw new RuntimeException("RPC timeout after " + timeoutSeconds + " seconds");
        }
        return response;
    }
}

public class RPCServer {

    public static void start(Channel channel) throws Exception {
        channel.queueDeclare("rpc-queue", false, false, false, null);
        channel.basicQos(1); // process one request at a time

        System.out.println("RPC Server waiting for requests...");

        channel.basicConsume("rpc-queue", false,
            (consumerTag, delivery) -> {
                String request = new String(delivery.getBody(), "UTF-8");
                String correlationId = delivery.getProperties().getCorrelationId();
                String replyTo = delivery.getProperties().getReplyTo();

                String response;
                try {
                    response = handleRequest(request);
                } catch (Exception e) {
                    response = "{\"error\": \"" + e.getMessage() + "\"}";
                }

                // Send the response back to the reply queue
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                    .correlationId(correlationId)
                    .build();

                channel.basicPublish("", replyTo, replyProps, response.getBytes());
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            },
            consumerTag -> {}
        );
    }

    private static String handleRequest(String request) {
        // business logic — process the request and return a response
        return "{\"status\": \"ok\", \"result\": \"processed: " + request + "\"}";
    }
}

Retries with Exponential Backoff #

When a message fails to process because of a transient error (the database is down, another service isn’t responding), you need retries with a delay — not an immediate requeue, which would create a busy loop.

RabbitMQ doesn’t have a built-in delay, but you can simulate one with TTL queues:

public class RetryWithBackoff {

    // Setup: create a queue hierarchy for delays
    public static void setupRetryQueues(Channel channel) throws Exception {
        String mainExchange = "order-exchange";
        String mainQueue = "order-queue";
        String retryExchange = "retry-exchange";

        // Main exchange
        channel.exchangeDeclare(mainExchange, "direct", true);

        // Retry queues with TTL — messages will "die" back to the main queue after the delay
        String[] retryQueues = {"retry-5s", "retry-30s", "retry-5m"};
        int[] delays = {5_000, 30_000, 300_000};

        for (int i = 0; i < retryQueues.length; i++) {
            java.util.Map<String, Object> args = new java.util.HashMap<>();
            args.put("x-message-ttl", delays[i]);
            args.put("x-dead-letter-exchange", mainExchange); // return to the main queue after the TTL
            args.put("x-dead-letter-routing-key", "order");

            channel.queueDeclare(retryQueues[i], true, false, false, args);
            channel.queueBind(retryQueues[i], retryExchange, retryQueues[i]);
        }

        // DLQ for messages that have exhausted their retries
        channel.queueDeclare("dlq.order-queue", true, false, false, null);

        // Main queue with a DLX
        java.util.Map<String, Object> mainArgs = new java.util.HashMap<>();
        mainArgs.put("x-dead-letter-exchange", retryExchange);
        channel.queueDeclare(mainQueue, true, false, false, mainArgs);
        channel.queueBind(mainQueue, mainExchange, "order");
    }

    // The consumer that decides which retry queue a message should go to
    public static void consumeWithRetry(Channel channel) throws Exception {
        channel.basicQos(1);

        channel.basicConsume("order-queue", false,
            (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                long deliveryTag = delivery.getEnvelope().getDeliveryTag();

                // Count how many times this message has been retried
                var headers = delivery.getProperties().getHeaders();
                int retryCount = getRetryCount(headers);

                try {
                    processMessage(message);
                    channel.basicAck(deliveryTag, false);

                } catch (Exception e) {
                    if (retryCount >= 3) {
                        // Already retried 3 times — send to the DLQ
                        forwardToDLQ(channel, delivery, message, e.getMessage());
                        channel.basicAck(deliveryTag, false); // ack so it isn't processed again
                    } else {
                        // Nack without requeue → goes to the DLX → retry queue
                        channel.basicNack(deliveryTag, false, false);
                    }
                }
            },
            consumerTag -> {}
        );
    }

    private static int getRetryCount(java.util.Map<String, Object> headers) {
        if (headers == null) return 0;
        var xDeath = headers.get("x-death");
        if (xDeath == null) return 0;
        // x-death contains a list of death records
        @SuppressWarnings("unchecked")
        var deathList = (java.util.List<?>) xDeath;
        return deathList.size();
    }

    private static void processMessage(String message) throws Exception {
        System.out.println("Processing: " + message);
    }

    private static void forwardToDLQ(Channel channel, com.rabbitmq.client.Delivery delivery,
                                      String message, String reason) throws Exception {
        System.err.println("Message entering DLQ after 3 retries: " + reason);
        channel.basicPublish("", "dlq.order-queue",
            MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    }
}

When to Use RabbitMQ and When Not To #

USE RABBITMQ WHEN:
  ✓ You need flexible content-based routing (topic exchanges)
  ✓ Task queues — jobs that must execute exactly once
  ✓ Request-reply / RPC patterns
  ✓ Moderate volume with high routing complexity
  ✓ You need priority queues (x-max-priority)
  ✓ Messages must be deleted after processing (no replay needed)
  ✓ Integration with protocols other than AMQP (STOMP, MQTT via plugins)

CONSIDER ALTERNATIVES WHEN:
  ✗ Very high volume (millions of messages/second) → Kafka fits better
  ✗ You need to replay old messages → Kafka has log retention
  ✗ Many independent consumers reading the same data → Kafka consumer groups
  ✗ Stream processing and real-time aggregation → Kafka Streams
  ✗ Audit logs that must never be deleted → Kafka with long retention

flowchart TD
    A{"Need flexible\nrouting?"} -- Yes --> B{"Tasks executed\nexactly once?"}
    A -- No --> C{"Very high\nvolume?"}

    B -- Yes --> RABBIT[RabbitMQ]
    B -- No --> D{"Need\nrequest-reply?"}

    D -- Yes --> RABBIT
    D -- No --> C

    C -- Yes --> KAFKA[Kafka]
    C -- No --> E{"Need\nreplay?"}

    E -- Yes --> KAFKA
    E -- No --> RABBIT

Summary #

The exchange is the router — producers don’t send directly to queues. Choose the right exchange type: direct for exact match, fanout for broadcasting, topic for pattern-based routing.
Always use durable=true for queues and exchanges in production, and deliveryMode=2 (persistent) for messages that must not be lost when the broker restarts.
Disable auto ack and call basicAck manually after successful processing — this prevents message loss if a consumer crashes mid-process.
Set the prefetch count (basicQos) for fair load distribution between consumers. Without prefetch, slow consumers pile up unprocessable messages.
The Dead Letter Exchange (DLX) is RabbitMQ’s official way to handle failed messages — configured at the queue level so nacked messages automatically go to the DLQ.
The request-reply pattern is RabbitMQ’s advantage over Kafka — use correlationId and replyTo (amq.rabbitmq.reply-to) for synchronous communication on top of the messaging layer.
Retries with exponential backoff can be implemented using TTL queues that return messages to the main queue after a certain delay.
RabbitMQ fits best for task queues, complex routing, and request-reply — not for high-volume event streaming or scenarios requiring message replay.

← Previous: Kafka Next: Amazon SQS →