Publisher Confirms in RabbitMQ — Complete Guide

In this tutorial, you will learn about Publisher Confirms in RabbitMQ. We cover key concepts, practical examples, and best practices to help you master this topic.

Publisher confirms enable reliable message publishing in RabbitMQ by acknowledging when the broker has received and processed each published message.

What You Learn

You will learn how publisher confirms work, how to implement them in Python with Pika, and how they differ from transactions for guaranteeing message delivery.

Why It Matters

Without publisher confirms, a producer has no way of knowing whether the broker received a message. Network failures, broker crashes, or channel errors can silently drop messages. Publisher confirms give producers certainty about delivery.

Real-World Use

Doda Browser uses publisher confirms when sending malware scan requests. If a confirm is not received within a timeout, the request is retried. This guarantees that no file is left unscanned due to messaging failures.

How Publisher Confirms Work

flowchart LR
    P[Producer] -->|publish| C[Channel]
    C -->|send| B[Broker]
    B -->|ack/nack| C
    C -->|callback| P
    style B fill:#f90,color:#fff

The producer publishes messages in confirm mode. The broker asynchronously sends confirmations. A basic.ack means the broker took ownership. A basic.nack means the broker could not Process the message.

Basic Implementation

Enable confirm mode on the channel before publishing:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.confirm_delivery()

channel.queue_declare(queue='confirmed_queue', durable=True)

try:
    channel.basic_publish(
        exchange='',
        routing_key='confirmed_queue',
        body='This message is confirmed',
        properties=pika.BasicProperties(delivery_mode=2),
        mandatory=True
    )
    print("Message confirmed by broker")
except pika.exceptions.UnroutableError:
    print("Message was not routable")
except pika.exceptions.NackError:
    print("Message was nacked by broker")

connection.close()

Expected output:

Message confirmed by broker

Batch Confirms

For high-throughput scenarios, publish multiple messages then wait for confirms:

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.confirm_delivery()

channel.queue_declare(queue='batch_queue', durable=True)

batch_size = 100
start = time.time()

for i in range(batch_size):
    channel.basic_publish(
        exchange='',
        routing_key='batch_queue',
        body=f'Message {i}',
        properties=pika.BasicProperties(delivery_mode=2)
    )

channel.wait_for_confirms()
elapsed = time.time() - start
print(f"Published {batch_size} messages in {elapsed:.2f}s")
print(f"Rate: {batch_size/elapsed:.0f} msgs/s")
channel.queue_delete(queue='batch_queue')
connection.close()

Expected output:

Published 100 messages in 0.15s
Rate: 666 msgs/s

Async Confirms with Callbacks

For non-blocking confirm handling:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.confirm_delivery()

acks = 0
nacks = 0

def on_confirm(frame):
    global acks, nacks
    if frame.method.NAME == 'Basic.Ack':
        acks += 1
    elif frame.method.NAME == 'Basic.Nack':
        nacks += 1
    print(f"Acks: {acks}, Nacks: {nacks}")

channel.add_on_confirm_callback(on_confirm)

channel.queue_declare(queue='async_confirm', durable=True)
channel.basic_publish(exchange='', routing_key='async_confirm', body='msg1')
channel.basic_publish(exchange='', routing_key='async_confirm', body='msg2')
channel.basic_publish(exchange='', routing_key='async_confirm', body='msg3')

connection.close()

Expected output:

Acks: 1, Nacks: 0
Acks: 2, Nacks: 0
Acks: 3, Nacks: 0

Publisher Confirms vs Transactions

Transactions provide atomicity but are much slower. Confirms are asynchronous and preferred.

Common Mistakes

1. Forgetting to Enable Confirm Mode

Without channel.confirm_delivery(), publishes are fire-and-forget. No error is raised if the broker rejects the message.

2. Using Confirm Mode with Transactions

You cannot use confirm mode and transactions on the same channel. They are mutually exclusive.

3. Blocking on Every Publish

Calling wait_for_confirms() after every single message kills throughput. Batch confirms or async callbacks are much faster.

4. Ignoring Nacks

A basic.nack means the broker could not process the message. Handle nacks by logging, alerting, or retrying.

5. Not Setting Delivery Mode

Publisher confirms verify the broker received the message but not that it was persisted. Set delivery_mode=2 for persistent messages.

Practice Questions

1. What is a publisher confirm?

A confirmation from the broker that it has received and processed a published message. Sent asynchronously via basic.ack.

2. How do you enable publisher confirms in Pika?

Call channel.confirm_delivery() before publishing. This puts the channel in confirm mode.

3. What is the difference between ack and nack?

Ack means the broker successfully processed the message. Nack means the broker could not process it and the message may be lost.

4. Why are publisher confirms preferred over transactions?

Confirms are asynchronous and have minimal performance overhead. Transactions block on every commit and reduce throughput by orders of magnitude.

Challenge

Write a producer that publishes 1000 messages with batch confirms, measures throughput, and retries any nacked messages up to 3 times before giving up.

FAQ

Mini Project: Reliable Publisher

import pika
import time
import json

class ReliablePublisher:
    def __init__(self):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.conn.channel()
        self.channel.confirm_delivery()
        self.channel.queue_declare(queue='reliable_queue', durable=True)
        self.retries = {}

    def publish(self, message, max_retries=3):
        msg_id = str(time.time())
        self.retries[msg_id] = 0
        self._publish_with_retry(msg_id, message, max_retries)

    def _publish_with_retry(self, msg_id, message, max_retries):
        try:
            self.channel.basic_publish(
                exchange='',
                routing_key='reliable_queue',
                body=json.dumps({'id': msg_id, 'data': message}),
                properties=pika.BasicProperties(delivery_mode=2, message_id=msg_id)
            )
            print(f"Published: {msg_id}")
            del self.retries[msg_id]
        except (pika.exceptions.UnroutableError, pika.exceptions.NackError):
            self.retries[msg_id] += 1
            if self.retries[msg_id] <= max_retries:
                print(f"Retry {self.retries[msg_id]}/{max_retries} for {msg_id}")
                self._publish_with_retry(msg_id, message, max_retries)
            else:
                print(f"Failed after {max_retries} retries: {msg_id}")
                del self.retries[msg_id]

pub = ReliablePublisher()
pub.publish('msg1')
pub.publish('msg2')
pub.publish('msg3')
pub.conn.close()

Expected output:

Published: 1687958400.123456
Published: 1687958400.123789
Published: 1687958400.124012

What's Next

Now that you understand publisher confirms, explore consumer acknowledgements for reliable message consumption, then learn about QoS prefetch for controlling consumer throughput.