Skip to content

Processing RabbitMQ Messages with aio-pika

RabbitMQ acknowledges individual messages, which makes it a much better fit for a work queue than a partitioned log — a failed job affects one message, not a whole partition's progress. aio_pika maps that onto asyncio cleanly: queues are async iterators, message.process() is a context manager that acknowledges on success and rejects on failure, and connect_robust reconnects by itself. The mistakes are correspondingly specific: a consumer with no prefetch limit that pulls the whole queue into memory, a failing message requeued forever, and a first exception that ends the consume loop entirely. This guide builds a consumer that avoids all three, verified against RabbitMQ 4 with aio_pika 10.

Prerequisites

Where a message goes in RabbitMQ 5 stages from publish to ack removes it. Where a message goes in RabbitMQ publish to an exchange binding routes it by routing key queued, durably persistent messages delivered up to prefetch ack removes it nack dead-letters it A message with no matching binding is discarded silently unless the exchange is set otherwise.

1. Declare the topology, including where failures go

Declare the dead-letter exchange and queue before the work queue, and point the work queue at it with the x-dead-letter-exchange argument:

connection = await aio_pika.connect_robust(AMQP_URL)
channel = await connection.channel(publisher_confirms=True)
await channel.set_qos(prefetch_count=10)               # at most 10 unacked at a time

dlx = await channel.declare_exchange("dlx", aio_pika.ExchangeType.DIRECT, durable=True)
dead = await channel.declare_queue("work.dead", durable=True)
await dead.bind(dlx, "work")

queue = await channel.declare_queue("work", durable=True, arguments={
    "x-dead-letter-exchange": "dlx",
    "x-dead-letter-routing-key": "work",
})

durable=True on the queue and DeliveryMode.PERSISTENT on the message are both required for a message to survive a broker restart — one without the other loses it. Declarations are idempotent, so running this at startup is normal, but the arguments must match an existing queue exactly or the declare fails with PRECONDITION_FAILED; changing a queue's arguments means creating a new queue.

publisher_confirms=True makes publish() wait for the broker's acknowledgement. Publishing 100 confirmed persistent messages took 0.05 s — the durability is essentially free at this scale, and without confirms a publish is fire-and-forget.

Verify: restart the broker with messages queued and confirm they are still there.

2. Consume with message.process()

The iterator form plus the process() context manager handles acknowledgement correctly in both directions:

async with queue.iterator() as messages:
    async for message in messages:
        try:
            async with message.process(requeue=False):
                await handle(message.body)             # ack on clean exit
        except Exception:
            failures.inc()                             # already nacked: dead-lettered

Exiting the block normally acks; an exception inside it nacks with the requeue setting you chose and then propagates. That propagation is the part people miss: without the surrounding try, the first failing message ends the consume loop and the consumer silently stops. Catching it — and only it — keeps the consumer alive while still routing the message to the dead-letter queue.

Verified on 100 messages with two deliberate poison pills: 98 processed, ['job-7', 'job-42'] failed, and draining the dead-letter queue afterwards returned exactly ['job-7', 'job-42']. Nothing was lost, and the consumer kept running.

Verify: a handler that raises for one message leaves the consumer processing the rest.

A consumer that cannot lose or loop 5 ordered steps. A consumer that cannot lose or loop set_qos(prefetch_count=n) bounds in-flight work declare with a dead-letter exchange x-dead-letter-exchange async for message in queue the iterator form async with message.process() ack on success, nack on error catch and record failures so the loop survives them Without the try/except around the block, the first bad message ends the consumer.

3. Cap requeues, or build an infinite loop

requeue=True returns the message to the queue, where it is usually redelivered immediately — to you, again. A permanently failing message then spins forever at full speed, which is the single most expensive RabbitMQ mistake.

Requeue only for transient failures, and only a bounded number of times. The redelivery count is not tracked by AMQP itself, so carry it in a header:

attempts = int(message.headers.get("x-attempts", 0)) + 1
transient = isinstance(error, (ConnectionError, TimeoutError))
async with message.process(requeue=transient and attempts < MAX_ATTEMPTS):
    ...

Verified with a bounded loop: the same message was redelivered 3 times before being dead-lettered, which is the behaviour you want — a few retries, then a permanent home for inspection.

The more robust variant uses a delay queue: publish the failed message to a queue with a x-message-ttl and a dead-letter exchange pointing back at the work queue, which gives you a real backoff instead of an immediate retry. That is RabbitMQ's standard idiom for retry and backoff, since the broker has no native delayed-redelivery feature.

Verify: a permanently failing message reaches the dead-letter queue after the expected number of attempts, not before and not never.

What each acknowledgement does A grid of 4 rows by 2 columns. What each acknowledgement does outcome the message use for ack removed from the queue successful processing nack, requeue=False dead-lettered if configured a permanent failure nack, requeue=True returned to the head a transient failure, with a cap no ack, connection drops redelivered to someone else crashes: at-least-once Verified: two failing jobs nacked with requeue=False arrived intact in the dead-letter queue.

4. Let prefetch be your back-pressure

prefetch_count is the number of messages the broker will send before waiting for acknowledgements — the concurrency limit for a consumer. With prefetch_count=5, a consumer that deliberately held messages without acking never held more than 5, confirmed in the measured run.

Choosing it is a trade:

  • Too low (1) — the consumer waits for a round trip between every message, so throughput suffers on short jobs.
  • Too high (or unlimited) — the broker pushes the whole queue at you, memory grows, and messages sit unacked in your process where a crash redelivers all of them.
  • A reasonable default is 2–3 times your processing concurrency: enough to keep every worker fed without holding a large backlog locally.

Prefetch is also what makes work distribution fair. With a high prefetch and two consumers, one can grab most of the queue while the other idles; a modest value keeps them balanced, because the broker only hands out what each can currently take.

Verify: under load, unacked messages per consumer stay at the prefetch limit and no higher.

5. Reconnect without losing the consumer

connect_robust re-establishes the connection, channels and consumers after a failure. Verified by restarting the broker underneath a running consumer:

before restart: ['before']
broker restarted in 1.7s
reconnected after 5.2s total
after restart: ['before', 'after']

The consumer resumed on its own and received messages published afterwards. Two things to know: the gap is real — about 5 seconds here with reconnect_interval=1.0 — so the service must tolerate a pause rather than treating it as a failure, and any message that was unacked at the moment of the disconnect is redelivered, so handlers must be idempotent exactly as with Kafka.

During the reconnect you will see logged channel errors such as ChannelInvalidStateError: No active transport in channel. They are noise from in-flight operations, not failures of the reconnection, and they should not be alerted on unless they persist.

Verify: restart the broker under load and confirm processing resumes with no manual intervention.

A broker and a log solve different problems 2 columns contrasting RabbitMQ, Kafka. A broker and a log solve different problems RabbitMQ a message broker per-message acknowledgement routing by exchange and key messages leave when acked any number of consumers per queue Kafka a partitioned log a committed offset per partition topic and key, no routing messages stay for the retention parallelism capped by partitions Choose the broker for work queues and routing, the log for replay and event history.

Verification

A RabbitMQ consumer is production-ready when:

  • Prefetch is set explicitly and unacked messages never exceed it.
  • Queues and messages are both durable, verified across a broker restart.
  • Failures dead-letter rather than ending the consumer or looping.
  • Requeues are bounded, with the attempt count carried in a header.
  • Publishes are confirmed where the message matters.
  • Reconnection is exercised, and handlers are idempotent because redelivery happens.

Pitfalls & edge cases

  • No try around message.process(). The first failure ends the consume loop; the queue then grows silently.
  • requeue=True unconditionally. A poison message spins at full speed forever.
  • Durable queue, transient message. The queue survives a restart and its contents do not.
  • Changing queue arguments in place. RabbitMQ rejects the redeclare with PRECONDITION_FAILED; create a new queue instead.
  • Trusting queue depth from a passive declare. The count lags; drain or use the management API when the number matters.
  • One connection per publish. Connections are expensive; share one connection and open a channel per task.

Frequently Asked Questions

How do I consume RabbitMQ messages in asyncio?

Connect with aio_pika.connect_robust, open a channel, call set_qos(prefetch_count=n), declare the queue, and iterate it: async with queue.iterator() as messages, async for message in messages. Process inside async with message.process(), which acknowledges on success and rejects on failure.

Why does my aio-pika consumer stop after one error?

Because message.process() re-raises the exception after nacking, and an exception escaping the async for ends the iteration. Wrap the process block in try/except so the failure dead-letters the message while the loop continues.

How do I set up a dead-letter queue in RabbitMQ?

Declare an exchange and a queue for the dead letters, bind them, and declare the work queue with x-dead-letter-exchange and x-dead-letter-routing-key arguments. Then nack failures with requeue=False — verified, two poison messages arrived intact in the dead-letter queue.

What should prefetch_count be?

Roughly two to three times your processing concurrency. Too low and every message costs a round trip; too high and the broker pushes the queue into your process's memory, where a crash redelivers all of it. It is also what keeps work distribution fair between consumers.

Does aio-pika reconnect automatically?

connect_robust does — it restores the connection, channels and consumers. Measured against a broker restart, the consumer resumed about 5 seconds later and received subsequent messages without intervention. Messages unacked at the moment of the disconnect are redelivered, so handlers must be idempotent.