Publisher Confirms#
Warning
This is a beta version of the port from official tutorial. Please when you found an error create issue or pull request for me.
Note
Using the aio-pika async Python client
Note
Prerequisites
This tutorial assumes RabbitMQ is installed and running on localhost on standard port (5672). In case you use a different host, port or credentials, connections settings would require adjusting.
Where to get help
If you’re having trouble going through this tutorial you can contact us through the mailing list.
Publisher confirms are a RabbitMQ extension to implement reliable publishing. When publisher confirms are enabled on a channel, messages the client publishes are confirmed asynchronously by the broker, meaning they have been taken care of on the server side.
Overview#
In this tutorial we’re going to use publisher confirms to make sure published messages have safely reached the broker. We will cover several strategies to using publisher confirms and explain their pros and cons.
Enabling Publisher Confirms on a Channel#
Publisher confirms are a RabbitMQ extension to the AMQP 0.9.1 protocol.
Publisher confirms are enabled at the channel level by setting the publisher_confirms parameter to True,
which is the default.
channel = await connection.channel(
publisher_confirms=True, # This is the default
)
Strategy #1: Publishing Messages Individually#
Let’s start with the simplest approach to publishing with confirms, that is, publishing a message and waiting synchronously for its confirmation:
# Sending the messages
for msg in get_messages_to_publish():
# Waiting for publisher confirmation with timeout for every message
await channel.default_exchange.publish(
Message(msg),
routing_key=queue.name,
timeout=5.0,
)
In the previous example we publish a message as usual and wait for its confirmation with the await keyword.
The await returns as soon as the message has been confirmed.
If the message is not confirmed within the timeout or if it is nack-ed (meaning the broker could not take care of it for
some reason), the await will throw an exception.
The on_return_raises parameter of aio_pika.connect() and connection.channel() controls this behaivior for if a mandatory
message is returned.
The handling of the exception usually consists in logging an error message and/or retrying to send the message.
Different client libraries have different ways to synchronously deal with publisher confirms, so make sure to read carefully the documentation of the client you are using.
This technique is very straightforward but also has a major drawback: it significantly slows down publishing, as the confirmation of a message blocks the publishing of all subsequent messages. This approach is not going to deliver throughput of more than a few hundreds of published messages per second. Nevertheless, this can be good enough for some applications.
Strategy #2: Publishing Messages in Batches#
To improve upon our previous example, we can publish a batch of messages and wait for this whole batch to be confirmed. The following example uses a batch of 100:
batchsize = 100
outstanding_messages = []
# Sending the messages
for msg in get_messages_to_publish():
outstanding_messages.append(
asyncio.create_task(
channel.default_exchange.publish(
Message(msg),
routing_key=queue.name,
timeout=5.0,
)
)
)
# Yield control flow to event loop, so message sending is initiated:
await asyncio.sleep(0)
if len(outstanding_messages) == batchsize:
await asyncio.gather(*outstanding_messages)
outstanding_messages.clear()
if len(outstanding_messages) > 0:
await asyncio.gather(*outstanding_messages)
outstanding_messages.clear()
Waiting for a batch of messages to be confirmed improves throughput drastically over waiting for a confirm for individual message (up to 20-30 times with a remote RabbitMQ node). One drawback is that we do not know exactly what went wrong in case of failure, so we may have to keep a whole batch in memory to log something meaningful or to re-publish the messages. And this solution is still synchronous, so it blocks the publishing of messages.
Note
To initiate message sending asynchronously, a task is created with asyncio.create_task, so the execution of our function
is handled by the event-loop.
The await asyncio.sleep(0) is required to make the event loop switch to our coroutine.
Any await would have sufficed, though.
Using async for with an async generator also requires the generator to yield control flow with await for message
sending to be initiated.
Without the task and the await the message sending would only be initiated with the asyncio.gather call.
For some applications this behaivior might be acceptable.
Strategy #3: Handling Publisher Confirms Asynchronously#
The broker confirms published messages asynchronously, our helper function will publish the messages and be notified of these confirms:
# List for storing tasks
tasks = []
# Sending the messages
for msg in get_messages_to_publish():
task = asyncio.create_task(
publish_and_handle_confirm(
channel.default_exchange,
queue.name,
msg,
)
)
tasks.append(task)
# Yield control flow to event loop, so message sending is initiated:
await asyncio.sleep(0)
# Await all tasks
await asyncio.gather(*tasks)
In Python 3.11 a TaskGroup can be used instead of the list with asyncio.gather.
The helper function publishes the message and awaits the confirmation. This way the helper function knows which message the confirmation, timeout or rejection belongs to.
async def publish_and_handle_confirm(exchange, queue_name, message_body):
try:
confirmation = await exchange.publish(
Message(message_body),
routing_key=queue_name,
timeout=5.0,
)
except DeliveryError as e:
print(f"Delivery of {message_body!r} failed with exception: {e}")
except TimeoutError:
print(f"Timeout occured for {message_body!r}")
else:
if not isinstance(confirmation, Basic.Ack):
print(f"Message {message_body!r} was not acknowledged by broker!")
Summary#
Making sure published messages made it to the broker can be essential in some applications. Publisher confirms are a RabbitMQ feature that helps to meet this requirement. Publisher confirms are asynchronous in nature but it is also possible to handle them synchronously. There is no definitive way to implement publisher confirms, this usually comes down to the constraints in the application and in the overall system. Typical techniques are:
publishing messages individually, waiting for the confirmation synchronously: simple, but very limited throughput.
publishing messages in batch, waiting for the confirmation synchronously for a batch: simple, reasonable throughput, but hard to reason about when something goes wrong.
asynchronous handling: best performance and use of resources, good control in case of error, but can be involved to implement correctly.
Note
This material was adopted from official tutorial on rabbitmq.org.