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 Go and rabbitmq/amqp091-go examples.
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.
producer → [single queue] → consumer 1
→ consumer 2
→ consumer 3
The broker distributes the tasks among the workers itself. Each message goes to exactly one of them.
ch.Qos(20, 0, false) // no more than 20 unacknowledged messages per consumer
q, err := ch.QueueDeclare(
"images.to-process",
true, // durable
false, // autoDelete
false, // exclusive
false, // noWait
amqp.Table{"x-queue-type": "quorum"},
)
msgs, err := ch.Consume(q.Name, "", false /* autoAck */, false, false, false, nil)
for i := 0; i < 20; i++ {
go func() {
for d := range msgs {
processImage(d.Body) // process the image
d.Ack(false)
}
}()
}
20 goroutines read from a single delivery channel, and Qos caps the number of unacknowledged messages. If you run 5 copies of the service, you get 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.
ch.ExchangeDeclare("cache.invalidation", "fanout", true, false, false, false, nil)
// the service's own queue: unnamed, exclusive, auto-delete
q, err := ch.QueueDeclare("", false, true, true, false, nil)
ch.QueueBind(q.Name, "", "cache.invalidation", false, nil)
msgs, err := ch.Consume(q.Name, "", true, true, false, false, nil)
for d := range msgs {
var event CacheInvalidationEvent
if err := json.Unmarshal(d.Body, &event); err != nil {
continue
}
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.
ch.ExchangeDeclare("orders", "direct", true, false, false, false, nil)
quorum := amqp.Table{"x-queue-type": "quorum"}
ch.QueueDeclare("orders.fulfillment", true, false, false, false, quorum)
ch.QueueDeclare("orders.audit", true, false, false, false, quorum)
ch.QueueDeclare("orders.alerts", true, false, false, false, quorum)
ch.QueueBind("orders.fulfillment", "order.created", "orders", false, nil)
ch.QueueBind("orders.audit", "order.created", "orders", false, nil)
ch.QueueBind("orders.audit", "order.cancelled", "orders", false, nil)
ch.QueueBind("orders.alerts", "order.payment-failed", "orders", false, nil)
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.
ch.ExchangeDeclare("events", "topic", true, false, false, false, nil)
quorum := amqp.Table{"x-queue-type": "quorum"}
ch.QueueDeclare("audit.orders", true, false, false, false, quorum)
ch.QueueDeclare("dashboard.eu", true, false, false, false, quorum)
ch.QueueDeclare("alerts.critical", true, false, false, false, quorum)
ch.QueueBind("audit.orders", "order.#", "events", false, nil)
ch.QueueBind("dashboard.eu", "*.*.eu", "events", false, nil)
ch.QueueBind("alerts.critical", "payment.failed.#", "events", false, nil)
A message with the key order.cancelled.eu will land in audit.orders (via order.#) and in dashboard.eu (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 amqp091-go the request-response is assembled by hand — with direct reply-to and CorrelationId:
// Client
func (c *PricingClient) Quote(ctx context.Context, req QuoteRequest) (PriceQuote, error) {
replies, err := c.ch.Consume("amq.rabbitmq.reply-to", "", true, false, false, false, nil)
if err != nil {
return PriceQuote{}, err
}
corrID := uuid.NewString()
body, _ := json.Marshal(req)
err = c.ch.PublishWithContext(ctx, "pricing.exchange", "pricing.quote", false, false,
amqp.Publishing{
CorrelationId: corrID,
ReplyTo: "amq.rabbitmq.reply-to",
Body: body,
})
if err != nil {
return PriceQuote{}, err
}
for {
select {
case d := <-replies:
if d.CorrelationId != corrID {
continue
}
var quote PriceQuote
err := json.Unmarshal(d.Body, "e)
return quote, err
case <-ctx.Done():
return PriceQuote{}, ctx.Err()
}
}
}
// Server
for d := range requests {
var req QuoteRequest
if err := json.Unmarshal(d.Body, &req); err != nil {
d.Nack(false, false)
continue
}
body, _ := json.Marshal(ComputePriceQuote(req))
ch.PublishWithContext(ctx, "", d.ReplyTo, false, false, // the reply goes to reply-to
amqp.Publishing{CorrelationId: d.CorrelationId, Body: body})
d.Ack(false)
}
Direct reply-to (amq.rabbitmq.reply-to) is a RabbitMQ pseudo-queue: the broker delivers the reply straight into the client's connection, without creating a temporary queue. The server publishes the reply to d.ReplyTo with the same CorrelationId.
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:
func (p *PaymentProcessor) Process(ctx context.Context, event PaymentEvent) error {
tx, err := p.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
res, err := tx.ExecContext(ctx,
`INSERT INTO processed_events (idempotency_key) VALUES ($1)
ON CONFLICT DO NOTHING`, event.IdempotencyKey)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return nil // already processed — just acknowledge receipt
}
if _, err := tx.ExecContext(ctx,
`UPDATE account SET balance = balance - $1 WHERE id = $2`,
event.Amount, event.AccountID); err != nil {
return err
}
return tx.Commit()
}
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:
func (s *OrderService) OnOrderConfirmed(ctx context.Context, event OrderConfirmedEvent) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
var status string
if err := tx.QueryRowContext(ctx,
`SELECT status FROM orders WHERE id = $1 FOR UPDATE`,
event.OrderID).Scan(&status); err != nil {
return err
}
if status == "CONFIRMED" {
return nil // already in the desired state
}
if _, err := tx.ExecContext(ctx,
`UPDATE orders SET status = 'CONFIRMED' WHERE id = $1`,
event.OrderID); err != nil {
return err
}
return tx.Commit()
}
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.
A delayed retry is assembled with x-message-ttl and a Dead Letter Exchange:
ch.QueueDeclare("orders.retry", true, false, false, false, amqp.Table{
"x-queue-type": "quorum",
"x-message-ttl": int32(30_000), // wait 30 seconds
"x-dead-letter-exchange": "orders",
"x-dead-letter-routing-key": "order.created",
})
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.
func (s *OrderService) Confirm(ctx context.Context, orderID string) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx,
`UPDATE orders SET status = 'CONFIRMED' WHERE id = $1`, orderID); err != nil {
return err
}
payload, _ := json.Marshal(OrderConfirmedEvent{OrderID: orderID})
if _, err := tx.ExecContext(ctx,
`INSERT INTO outbox (id, routing_key, exchange, payload)
VALUES ($1, $2, $3, $4)`,
uuid.NewString(), "order.confirmed", "orders", payload); err != nil {
return err
}
return tx.Commit()
}
// Background publishing — a goroutine with a ticker
func (s *OrderService) PublishOutbox(ctx context.Context) {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
batch, err := s.outbox.FetchUnpublished(ctx, 100)
if err != nil {
continue
}
for _, e := range batch {
if err := s.ch.PublishWithContext(ctx, e.Exchange, e.RoutingKey, false, false,
amqp.Publishing{Body: e.Payload}); err != nil {
break
}
s.outbox.MarkPublished(ctx, e.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
| Task | Pattern | Exchange type |
|---|---|---|
| Distribute load across workers | Work Queue | direct (default) |
| Broadcast events to all services | Publish/Subscribe | fanout |
| Different events to different queues | Routing | direct |
| Pattern subscription to hierarchical events | Topic | topic |
| Synchronous call over a queue | RPC | direct + reply-to |
| Protection against redelivery | Idempotent Consumer | any |
| Retry with a delay | Delayed Retry | direct + DLX |
| Atomic publishing together with a DB write | Outbox | direct |
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-toandcorrelation-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.