aio-pika Documentation

aio-pika is an asynchronous Python client for RabbitMQ built on top of aiormq. It provides an object-oriented API with the following key features:

• Transparent auto-reconnects: Includes full state recovery.
• Publisher confirms: Support for RabbitMQ publisher confirms.
• Transactions: Support for AMQP transactions.
• Type-hints: Complete type-hints coverage for better developer experience.

While a single connect_robust() connection is usually sufficient due to AMQP multiplexing, connection pooling can be useful if:

• A single TCP connection becomes saturated by high message volume.
• You need to isolate groups of channels to prevent slow connections from affecting other workloads.
• Many concurrent tasks are contending for the same connection's write lock.

Use aio_pika.pool.Pool to manage a pool of connections or channels.

You can change how data is serialized and deserialized within the Master/Worker or RPC patterns by inheriting from the pattern base classes and redefining the following methods:

• aio_pika.patterns.base.serialize
• aio_pika.patterns.base.deserialize

Understanding the basic messaging model in RabbitMQ:

• Producer: A program that sends messages (the 'sending' action).
• Consumer: A program that waits to receive messages (the 'receiving' action).
• Queue: A mailbox inside RabbitMQ where messages are stored. It acts as an infinite buffer. Many producers can send to one queue, and many consumers can receive from one queue.
• Exchange: In RabbitMQ, messages are never sent directly to a queue; they always go through an exchange. A 'default exchange' (identified by an empty string "") allows you to route a message to a specific queue by specifying the queue name as the routing_key.

In RabbitMQ, producers do not send messages directly to queues. Instead, they send messages to an Exchange. The exchange receives messages and decides how to push them to queues based on its type.

Available exchange types (found in aio_pika.ExchangeType) include:

• DIRECT
• TOPIC
• HEADERS
• FANOUT (broadcasts all messages to all queues it knows)

To use a named exchange, you must declare it before publishing. Publishing to a non-existing exchange is forbidden.

To prevent message loss when a worker dies mid-task, use message acknowledgments. An acknowledgment (ack) tells RabbitMQ that a message has been processed and can be safely deleted. If a consumer's connection is lost before sending an ack, RabbitMQ will re-queue the message and deliver it to another available consumer.

Warning: Forgetting to call ack() will cause RabbitMQ to keep messages in an 'unacknowledged' state, leading to increased memory usage on the server. You can debug this using rabbitmqctl list_queues name messages_ready messages_unacknowledged.

async def on_message(message: IncomingMessage):
    print(" [x] Received %r" % message.body)
    await asyncio.sleep(message.body.count(b'.'))
    print(" [x] Done")
    await message.ack()

A Topic exchange routes messages to queues based on matching a routing_key against a binding_key.

Routing Key Rules:

• Must be a list of words delimited by dots (e.g., "stock.usd.nyse", "quick.orange.rabbit").
• The number of words can vary, but the exchange logic depends on the binding patterns used.

Binding Key Special Characters:

• * (star): Substitutes for exactly one word.
• # (hash): Substitutes for zero or more words.

Behavioral Notes:

• If a queue is bound with #, it acts like a fanout exchange (receives everything).
• If no special characters (* or #) are used in bindings, it behaves like a direct exchange.
• If a message's routing key does not match any bindings, it is discarded.

A binding is a relationship between an exchange and a queue, indicating that a queue is interested in messages from that exchange.

When creating a binding, you can provide a routing_key (often called a binding key in this context). The behavior of this key depends on the exchange type:

• Fanout exchanges: Ignore the routing key and broadcast messages to all bound queues.
• Direct exchanges: Use the routing key to filter messages. A message is only routed to queues whose binding key exactly matches the message's routing key.