Cheat SheetsRabbitMQMessaging Patterns

Messaging Patterns — Cheat Sheet

RabbitMQ · 4 topics. Download the PDF or the Instagram carousel and share it.

Cheat Sheet · AiCanCode.org
Messaging Patterns
RabbitMQ4 topicsQuick revision reference
1

Work Queues (Task Distribution)

Multiple workers consume from a single queue; with prefetch=1 and manual ack, work is evenly distributed to idle workers rather than buffered on a busy one.

  • ✓Work queues distribute tasks across multiple competing workers — each task is delivered to exactly one worker.
  • ✓prefetch=1 (basicQos) + manual ack = fair dispatch: a busy worker does not receive new tasks until it acks the current one.
  • ✓Without prefetch=1, RabbitMQ round-robins messages regardless of worker speed, causing uneven load distribution.
  • ✓Durable queue + persistent messages (deliveryMode=2) ensures no task loss on broker restart.
  • ✓Scale workers horizontally by increasing Kubernetes replicas or @RabbitListener concurrency; queue depth is the scaling signal.
  • ✓KEDA can autoscale K8s Deployments based on RabbitMQ queue depth — scale to zero when queue is empty.
Java — Work Queue Producer + Worker
// Producer — create work and publish
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection conn = factory.newConnection();
     Channel channel = conn.createChannel()) {

    // Durable queue — survives restart
    channel.queueDeclare("task_queue", true, false, false, null);

    String[] tasks = {"resize_image_1", "send_email_2", "process_payment_3"};
    for (String task : tasks) {
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .deliveryMode(2)  // persistent — survives restart
            .build();
        channel.basicPublish("", "task_queue", props, task.getBytes());
        System.out.println("Sent: " + task);
    }
}

// Worker — multiple instances of this can run in parallel
try (Connection conn = factory.newConnection();
     Channel channel = conn.createChannel()) {

    channel.queueDeclare("task_queue", true, false, false, null);
    channel.basicQos(1);  // fair dispatch — max 1 unacked message at a time

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String task = new String(delivery.getBody());
        System.out.println("Processing: " + task);
        try {
            doWork(task);  // potentially slow
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  // ack on success
        } catch (Exception e) {
            channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); // requeue on failure
        }
    };

    channel.basicConsume("task_queue", false, deliverCallback, tag -> {});
}
2

Publish-Subscribe Pattern

Fanout exchange broadcasts events to all bound queues; each consumer group has its own queue, receiving every event independently for fan-out notification scenarios.

  • ✓Fanout exchange ignores routing keys and delivers to all bound queues — ideal for broadcasting domain events
  • ✓Each subscriber gets its own queue; multiple instances of the same subscriber share one queue for load balancing
  • ✓Topic exchanges extend pub/sub with wildcard routing keys: * (one word) and # (zero or more words)
  • ✓Producers are decoupled from consumers — new subscribers can be added without any publisher changes
  • ✓Use durable queues + persistent messages for subscribers that must not miss events during restart
  • ✓Auto-delete queues (autoDelete=true) are useful for transient subscribers like live dashboard WebSocket feeds
Java — Spring AMQP fanout exchange with multiple subscriber queues
// Spring AMQP fanout pub/sub configuration
@Configuration
public class PubSubConfig {

    @Bean
    public FanoutExchange orderEventsExchange() {
        return new FanoutExchange("order.events", true, false);
    }

    @Bean public Queue emailNotificationQueue() {
        return QueueBuilder.durable("order.events.email").build();
    }
    @Bean public Queue inventoryUpdateQueue() {
        return QueueBuilder.durable("order.events.inventory").build();
    }
    @Bean public Queue analyticsQueue() {
        return QueueBuilder.durable("order.events.analytics").build();
    }

    // Bind all queues to the exchange — routing key ignored for fanout
    @Bean public Binding emailBinding(FanoutExchange ex, Queue emailNotificationQueue) {
        return BindingBuilder.bind(emailNotificationQueue).to(ex);
    }
    @Bean public Binding inventoryBinding(FanoutExchange ex, Queue inventoryUpdateQueue) {
        return BindingBuilder.bind(inventoryUpdateQueue).to(ex);
    }
    @Bean public Binding analyticsBinding(FanoutExchange ex, Queue analyticsQueue) {
        return BindingBuilder.bind(analyticsQueue).to(ex);
    }
}
3

Request-Reply (RPC) Pattern

Producer sends a message with a reply-to queue and correlation-id; consumer processes the request and publishes the response to the reply-to queue; producer correlates the reply.

  • ✓RPC over AMQP uses replyTo (temporary queue) and correlationId to match requests to responses
  • ✓Spring AMQP's convertSendAndReceive() handles reply-to queue creation and correlation automatically
  • ✓Direct Reply-To (amq.rabbitmq.reply-to) avoids creating a new queue per client
  • ✓Always set a replyTimeout — a missing or crashed server will never send a reply
  • ✓Synchronous RPC over messaging couples services; prefer async callbacks for resilience
  • ✓The server must publish the response with the same correlationId from the request
Java — manual RPC over AMQP
// Client side
String replyQueue = channel.queueDeclare("", false, true, true, null).getQueue();
Map<String, CompletableFuture<byte[]>> pending = new ConcurrentHashMap<>();

channel.basicConsume(replyQueue, true, (tag, msg) -> {
    String corrId = msg.getProperties().getCorrelationId();
    CompletableFuture<byte[]> future = pending.remove(corrId);
    if (future != null) future.complete(msg.getBody());
}, tag -> {});

// Send request
String corrId = UUID.randomUUID().toString();
CompletableFuture<byte[]> result = new CompletableFuture<>();
pending.put(corrId, result);

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .correlationId(corrId)
        .replyTo(replyQueue)
        .build();
channel.basicPublish("", "rpc_queue", props, "42".getBytes());

byte[] response = result.get(5, TimeUnit.SECONDS);

// Server side
channel.basicConsume("rpc_queue", false, (tag, msg) -> {
    byte[] body = processRequest(msg.getBody());
    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
            .correlationId(msg.getProperties().getCorrelationId())
            .build();
    channel.basicPublish("", msg.getProperties().getReplyTo(), replyProps, body);
    channel.basicAck(msg.getEnvelope().getDeliveryTag(), false);
}, tag -> {});
4

Competing Consumers Pattern

Multiple consumer instances all subscribe to the same queue; the broker delivers each message to exactly one consumer, providing natural horizontal scaling for processing.

  • ✓Competing consumers deliver each message to exactly one worker — natural horizontal scaling
  • ✓prefetch=1 + manual ack ensures fair dispatch; workers only receive work they can handle
  • ✓RabbitMQ round-robins messages across consumers; slow workers won't block fast ones with prefetch=1
  • ✓Crashed workers: unacked messages are re-queued and delivered to another worker
  • ✓Competing consumers break message ordering — use consistent hashing if order matters
  • ✓Design workers to be idempotent — re-delivery after crash must be safe to re-process
Spring AMQP — competing consumers with manual ack
// Worker (same code, deployed N times)
@Component
class OrderWorker {

    @RabbitListener(
        queues = "order-processing",
        containerFactory = "workerContainerFactory"
    )
    public void process(Order order, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            orderService.fulfil(order);
            channel.basicAck(tag, false);           // done
        } catch (RecoverableException e) {
            channel.basicNack(tag, false, true);    // requeue for retry
        } catch (Exception e) {
            channel.basicNack(tag, false, false);   // → DLX, no requeue
        }
    }
}

@Bean
SimpleRabbitListenerContainerFactory workerContainerFactory(ConnectionFactory cf) {
    var f = new SimpleRabbitListenerContainerFactory();
    f.setConnectionFactory(cf);
    f.setPrefetchCount(1);                          // fair dispatch
    f.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return f;
}
Learn this free with Aria, your AI tutor → AiCanCode.org/learn/rabbitmq