Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
View this page
Edit this page
Toggle table of contents sidebar

# 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: AbstractExchange,
        queue_name: str,
        message_body: bytes,
    ) -> None:
        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.