Messaging Patterns with AMQP

← Back to the section

Once services start talking through a broker, a question arises: how do you correctly organize queues, exchanges, and subscriptions? Most tasks fit into a handful of standard schemes. Let's walk through each one with Spring AMQP examples.

The difference between them is not in the producer's code, but in how many receivers see a single publication.

Hand out tasks to several workers — Work Queue

Picture this: users upload photos, and each one has to be compressed and cropped into several sizes. That takes time. Doing it right inside the HTTP request is a no-go — the user would be waiting minutes.

The solution is a work queue: save the job into a queue and hand it to one of a pool of workers.

The broker distributes the tasks among the workers itself. Each message goes to exactly one of them.

@Configuration
class ImageProcessingTopology {
    @Bean
    Queue imagesQueue() {
        return QueueBuilder.durable("images.to-process").quorum().build();
    }
}

@Component
class ImageProcessor {
    @RabbitListener(queues = "images.to-process", concurrency = "5-20")
    public void process(ImageJob job) {
        // process the image
    }
}

concurrency = "5-20" means: at least 5 threads, at most 20. If you run 5 copies of the service, you get 25–100 parallel workers.

When to pick it: background tasks — file processing, sending emails, generating reports, anything that is "drop it in a queue and someone will pick it up".

Send an event to every service at once — Publish/Subscribe

A different task: the event "configuration updated" happened, and every service must refresh its cache. You can't know in advance who exactly is subscribed and how many services are running.

This is publish/subscribe: the publisher sends a single message, and all subscribers receive a copy at the same time.

Here you need a fanout exchange — it copies every message into all bound queues. Each service declares its own queue and binds it to the shared exchange.

@Configuration
class CacheInvalidationTopology {

    @Bean FanoutExchange cacheInvalidation() {
        return new FanoutExchange("cache.invalidation", true, false);
    }

    @Bean Queue serviceACache() {
        return QueueBuilder.nonDurable().exclusive().autoDelete().build();
    }

    @Bean Binding bindA(Queue serviceACache, FanoutExchange cacheInvalidation) {
        return BindingBuilder.bind(serviceACache).to(cacheInvalidation);
    }
}

@Component
class ServiceACacheListener {
    @RabbitListener(queues = "#{serviceACache.name}")
    public void invalidate(CacheInvalidationEvent event) {
        cache.evict(event.key());
    }
}

exclusive + autoDelete — the queue belongs to a single connection and is deleted when it disconnects. On a service restart, no garbage piles up in the broker.

When to pick it: cache invalidation, broadcast notifications to the whole cluster, configuration updates.

Route an event to the right worker — Routing

Sometimes you don't want "everyone", you want "exactly the one who needs it". For example: the order.created event should go to the fulfillment service and to audit, while order.payment-failed should go only to alerts.

This is routing: a direct exchange looks at the message's routing key and delivers it only to queues with a matching binding key.

@Configuration
class OrderRoutingTopology {

    @Bean DirectExchange orders() { return new DirectExchange("orders", true, false); }

    @Bean Queue fulfillment() { return QueueBuilder.durable("orders.fulfillment").quorum().build(); }
    @Bean Queue audit()       { return QueueBuilder.durable("orders.audit").quorum().build(); }
    @Bean Queue alerts()      { return QueueBuilder.durable("orders.alerts").quorum().build(); }

    @Bean Binding b1(Queue fulfillment, DirectExchange orders) {
        return BindingBuilder.bind(fulfillment).to(orders).with("order.created");
    }
    @Bean Binding b2(Queue audit, DirectExchange orders) {
        return BindingBuilder.bind(audit).to(orders).with("order.created");
    }
    @Bean Binding b3(Queue audit, DirectExchange orders) {
        return BindingBuilder.bind(audit).to(orders).with("order.cancelled");
    }
    @Bean Binding b4(Queue alerts, DirectExchange orders) {
        return BindingBuilder.bind(alerts).to(orders).with("order.payment-failed");
    }
}

• order.created → fulfillment + audit.
• order.cancelled → audit only.
• order.payment-failed → alerts only.

When to pick it: explicit separation of flows — alerts apart from audit, the main worker apart from monitoring.

Subscribe by a pattern — Topic

Routing is great for strict rules. But what if a service wants to subscribe to "all order events"? Or "everything from the EU region"?

A topic exchange lets you define subscriptions with patterns. Message keys are built with dots (order.created.eu), and in a subscription you can use:

• * — exactly one word,
• # — zero or more words.

@Configuration
class TopicRoutingTopology {
    @Bean TopicExchange events() { return new TopicExchange("events", true, false); }

    @Bean Queue auditAllOrders() { return QueueBuilder.durable("audit.orders").quorum().build(); }
    @Bean Queue euDashboard()    { return QueueBuilder.durable("dashboard.eu").quorum().build(); }
    @Bean Queue alerts()         { return QueueBuilder.durable("alerts.critical").quorum().build(); }

    @Bean Binding b1(Queue auditAllOrders, TopicExchange events) {
        return BindingBuilder.bind(auditAllOrders).to(events).with("order.#");
    }
    @Bean Binding b2(Queue euDashboard, TopicExchange events) {
        return BindingBuilder.bind(euDashboard).to(events).with("*.*.eu");
    }
    @Bean Binding b3(Queue alerts, TopicExchange events) {
        return BindingBuilder.bind(alerts).to(events).with("payment.failed.#");
    }
}

A message with the key order.cancelled.eu will land in auditAllOrders (via order.#) and in euDashboard (via *.*.eu).

When to pick it: events with a hierarchical structure, when you need to subscribe flexibly without reworking the topology every time a new event type is added.

Request-response over a queue — RPC

Sometimes you need a synchronous response, but HTTP won't do: the service is behind NAT, has no public address, or you want load balancing across a pool of workers.

RPC over a queue: the client sends a request and waits for a response. The broker delivers the request to one of the workers, which replies to a separate reply queue. A correlation-id is used to match the request with the response.

In Spring AMQP this is hidden behind sendAndReceive:

// Client
@Component
@RequiredArgsConstructor
class PricingClient {
    private final RabbitTemplate rabbit;

    public PriceQuote quote(QuoteRequest request) {
        return (PriceQuote) rabbit.convertSendAndReceive(
            "pricing.exchange", "pricing.quote", request);
    }
}

// Server
@Component
class PricingServer {
    @RabbitListener(queues = "pricing.quote")
    public PriceQuote handle(QuoteRequest request) {
        return PriceQuote.compute(request); // the return value automatically goes to reply-to
    }
}

Spring AMQP creates a temporary reply queue itself, sets reply-to and correlation-id, and waits for the response. The return value from @RabbitListener is published back automatically.

When to pick it: you need a synchronous call, but HTTP doesn't work (NAT, firewall, no public address); you need to balance requests across a pool of workers.

When not to pick it: if HTTP/gRPC simply works — RPC over a broker is harder to debug and more expensive.

What to do about redelivery — Idempotent Consumer

AMQP guarantees at-least-once delivery: the same message may arrive twice. This happens when the broker didn't receive an acknowledgment for the processing (for example, because of a network issue) and re-sends the message.

A consumer must be able to handle repeats without breaking business logic.

Idempotency key in the database

The most reliable approach is to remember already-processed messages:

@RabbitListener(queues = "payments")
@Transactional
public void process(PaymentEvent event) {
    if (processedEventsRepo.existsByIdempotencyKey(event.idempotencyKey())) {
        return; // already processed — just acknowledge receipt
    }
    processedEventsRepo.save(new ProcessedEvent(event.idempotencyKey()));
    accountRepo.debit(event.accountId(), event.amount());
}

A processed_events table with a unique index on idempotency_key. If two identical messages arrive at the same time, the database catches the duplicate through a unique constraint violation.

Checking the object's state

If the event moves an object into a new state, it's enough to check the current one:

@Transactional
public void onOrderConfirmed(OrderConfirmedEvent event) {
    var order = orderRepo.findById(event.orderId()).orElseThrow();
    if (order.status() == OrderStatus.CONFIRMED) {
        return; // already in the desired state
    }
    order.confirm();
    orderRepo.save(order);
}

This needs no separate table — the state is already stored in the business object.

Retry with a delay and a Dead Letter Queue

What if the consumer failed not because of a bug, but because an external service was temporarily unavailable? You want to try again, but not immediately.

In Spring AMQP a delayed retry is assembled with x-message-ttl and a Dead Letter Exchange:

@Bean Queue retryQueue() {
    return QueueBuilder.durable("orders.retry")
        .withArgument("x-message-ttl", 30_000)             // wait 30 seconds
        .withArgument("x-dead-letter-exchange", "orders")
        .withArgument("x-dead-letter-routing-key", "order.created")
        .quorum().build();
}

The flow: the consumer rejects the message → it lands in the retry queue → after 30 seconds, once the TTL expires, it goes back through the DLX into the main queue → a new attempt.

The number of attempts is counted via the x-death.count header — you have to check it manually; there is no built-in limiter.

Messages that couldn't be processed after all attempts go to a Dead Letter Queue (DLQ) — a separate queue for manual inspection or alerts.

Guaranteed publishing — Outbox

Here's a common task: save an order to the database and publish an event — atomically. If you save first and publish afterward, the service may crash between the two operations. The event is lost.

The Outbox pattern: the event is saved in the same transaction as the business data. A separate process reads the table and publishes to AMQP.

@Transactional
public void confirm(OrderId orderId) {
    var order = orderRepo.findById(orderId).orElseThrow();
    order.confirm();
    orderRepo.save(order);
    outboxRepo.save(new OutboxEvent(
        UUID.randomUUID(),
        "order.confirmed",
        "orders",
        toJson(new OrderConfirmedEvent(orderId))
    ));
}

@Scheduled(fixedDelay = 500)
@Transactional
public void publishOutbox() {
    var batch = outboxRepo.fetchUnpublished(100);
    for (var event : batch) {
        rabbit.convertAndSend(event.exchange(), event.routingKey(), event.payload());
        outboxRepo.markPublished(event.id());
    }
}

Either both changes are committed, or neither is. Duplicates are possible (the publish succeeded, but marking it as sent didn't finish in time) — which is why the receiver still has to be idempotent.

Selection cheat sheet

In short

• Work Queue — one queue, several workers, each message goes to exactly one. For background tasks.
• Publish/Subscribe — a fanout exchange copies the message into all bound queues. For broadcast events.
• Routing — a direct exchange looks at the routing key. For precise separation of flows.
• Topic — like routing, but with * and # patterns. For hierarchical events with flexible subscription.
• RPC over a queue — request-response through the broker with reply-to and correlation-id. For calls without HTTP.
• Idempotent Consumer — at-least-once means possible duplicates. Protection: an idempotency key in the DB or a check of the object's state.
• Delayed Retry — TTL + DLX: the message is "parked" for a while, then comes back.
• Outbox — the event is saved in the same transaction as the data. Atomicity without a two-phase commit.

What to read next

• The AMQP protocol — the exchange/binding/queue model from the inside.
• Spring AMQP — configuration, RabbitTemplate, annotations.
• RabbitMQ in production — Quorum Queues, clustering, monitoring.
• AMQP vs Kafka — which broker to pick and when.