Messaging Patterns with AMQP in Go

← 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 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.

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, &quote)
			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

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.