Messaging Patterns with AMQP in Python

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

channel = await connection.channel()
await channel.set_qos(prefetch_count=20)

images_queue = await channel.declare_queue(
    "images.to-process",
    durable=True,
    arguments={"x-queue-type": "quorum"},
)

async def process(message: AbstractIncomingMessage) -> None:
    async with message.process():
        job = deserialize(message.body)
        # process the image

await images_queue.consume(process)

prefetch_count=20 means: the broker hands one consumer up to 20 unacknowledged messages at a time. If you run 5 copies of the service, you get up to 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.

cache_invalidation = await channel.declare_exchange(
    "cache.invalidation", ExchangeType.FANOUT, durable=True
)

service_a_cache = await channel.declare_queue(exclusive=True, auto_delete=True)
await service_a_cache.bind(cache_invalidation)

async def invalidate(message: AbstractIncomingMessage) -> None:
    async with message.process():
        event = deserialize(message.body)
        cache.evict(event.key)

await service_a_cache.consume(invalidate)

exclusive=True, auto_delete=True — 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.

orders = await channel.declare_exchange("orders", ExchangeType.DIRECT, durable=True)

quorum = {"x-queue-type": "quorum"}
fulfillment = await channel.declare_queue("orders.fulfillment", durable=True, arguments=quorum)
audit = await channel.declare_queue("orders.audit", durable=True, arguments=quorum)
alerts = await channel.declare_queue("orders.alerts", durable=True, arguments=quorum)

await fulfillment.bind(orders, routing_key="order.created")
await audit.bind(orders, routing_key="order.created")
await audit.bind(orders, routing_key="order.cancelled")
await alerts.bind(orders, routing_key="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.

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.

events = await channel.declare_exchange("events", ExchangeType.TOPIC, durable=True)

quorum = {"x-queue-type": "quorum"}
audit_all_orders = await channel.declare_queue("audit.orders", durable=True, arguments=quorum)
eu_dashboard = await channel.declare_queue("dashboard.eu", durable=True, arguments=quorum)
alerts = await channel.declare_queue("alerts.critical", durable=True, arguments=quorum)

await audit_all_orders.bind(events, routing_key="order.#")
await eu_dashboard.bind(events, routing_key="*.*.eu")
await alerts.bind(events, routing_key="payment.failed.#")

A message with the key order.cancelled.eu will land in audit_all_orders (via order.#) and in eu_dashboard (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 aio-pika this is hidden behind the RPC pattern:

from aio_pika.patterns import RPC

# Client
rpc = await RPC.create(channel)
quote = await rpc.call("pricing.quote", kwargs={"request": request})

# Server
async def handle(*, request: QuoteRequest) -> PriceQuote:
    return PriceQuote.compute(request)  # the return value automatically goes to reply-to

rpc = await RPC.create(channel)
await rpc.register("pricing.quote", handle, auto_delete=True)

aio-pika creates a temporary reply queue itself, sets reply-to and correlation-id, and waits for the response. The return value from the registered handler 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:

async def process(message: AbstractIncomingMessage) -> None:
    async with message.process():
        event = deserialize(message.body)
        async with session.begin():
            if await processed_events_repo.exists(event.idempotency_key):
                return  # already processed — just acknowledge receipt
            await processed_events_repo.save(ProcessedEvent(event.idempotency_key))
            await account_repo.debit(event.account_id, 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:

async def on_order_confirmed(event: OrderConfirmedEvent) -> None:
    async with session.begin():
        order = await order_repo.get(event.order_id)
        if order.status == OrderStatus.CONFIRMED:
            return  # already in the desired state
        order.confirm()
        await order_repo.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.

A delayed retry is assembled with x-message-ttl and a Dead Letter Exchange:

retry_queue = await channel.declare_queue(
    "orders.retry",
    durable=True,
    arguments={
        "x-message-ttl": 30_000,  # wait 30 seconds
        "x-dead-letter-exchange": "orders",
        "x-dead-letter-routing-key": "order.created",
        "x-queue-type": "quorum",
    },
)

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.

async def confirm(order_id: OrderId) -> None:
    async with session.begin():
        order = await order_repo.get(order_id)
        order.confirm()
        await order_repo.save(order)
        await outbox_repo.save(OutboxEvent(
            id=uuid.uuid4(),
            routing_key="order.confirmed",
            exchange="orders",
            payload=to_json(OrderConfirmedEvent(order_id)),
        ))

async def publish_outbox() -> None:  # background task
    while True:
        async with session.begin():
            batch = await outbox_repo.fetch_unpublished(100)
            for event in batch:
                exchange = await channel.get_exchange(event.exchange)
                await exchange.publish(
                    Message(event.payload.encode()), routing_key=event.routing_key
                )
                await outbox_repo.mark_published(event.id)
        await asyncio.sleep(0.5)

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