Consumer Acknowledgements in RabbitMQ — Complete Guide

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

Consumer acknowledgements let RabbitMQ know a message was processed successfully, enabling at-least-once delivery and preventing message loss.

What You Learn

You will learn how manual and automatic acknowledgement modes work, how to handle failures with nacks and requeueing, and best practices for reliable consumption.

Why It Matters

When a consumer receives a message but crashes before processing it, the message is lost if not acknowledged. Acknowledgements give the broker the information it needs to redeliver messages on failure, ensuring no data is lost.

Real-World Use

Durga Antivirus Pro uses manual acknowledgements for file scanning workers. If a worker crashes mid-scan, the message is requeued and picked up by another worker. This guarantees every file is scanned even if workers fail.

How Consumer Acknowledgements Work

flowchart LR
    B[Broker] -->|deliver| C[Consumer]
    C -->|basic.ack| B
    B -->|remove from queue| Q[Queue]
    C -->|basic.nack| B
    B -->|requeue or discard| Q
    style B fill:#f90,color:#fff

The broker delivers a message to a consumer. The consumer processes it then sends a basic.ack. If the consumer disconnects without acking, the broker requeues the message.

Automatic vs Manual Acknowledgement

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='ack_demo', durable=True)

# Auto-ack: broker removes message as soon as it is delivered
ch.basic_publish(exchange='', routing_key='ack_demo', body='auto_ack')
method_frame, _, body = ch.basic_get(queue='ack_demo', auto_ack=True)
print(f"Auto-ack received: {body.decode()}")
print("Message removed from queue on delivery")

conn.close()

Expected output:

Auto-ack received: auto_ack
Message removed from queue on delivery

Manual Acknowledgement

import pika
import time

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='manual_ack', durable=True)

ch.basic_publish(exchange='', routing_key='manual_ack', body='process_me')

def callback(ch, method, properties, body):
    print(f"Processing: {body.decode()}")
    time.sleep(0.5)
    print("Processing complete")
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print("Ack sent to broker")

ch.basic_consume(queue='manual_ack', on_message_callback=callback, auto_ack=False)
print("Waiting for messages...")
ch.start_consuming()

Expected output:

Processing: process_me
Processing complete
Ack sent to broker

Negative Acknowledgement (Nack)

Use basic.nack to reject a message that could not be processed:

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='nack_demo', durable=True)

ch.basic_publish(exchange='', routing_key='nack_demo', body='bad_message')

def callback(ch, method, properties, body):
    print(f"Received: {body.decode()}")
    print("Processing failed, sending nack with requeue=false")
    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    print("Message discarded")

ch.basic_consume(queue='nack_demo', on_message_callback=callback, auto_ack=False)
ch.start_consuming()

Expected output:

Received: bad_message
Processing failed, sending nack with requeue=false
Message discarded

Requeue vs Discard

When nacking, set requeue=True to return the message to the queue, or requeue=False to discard or send to a dead letter exchange:

# Requeue: message goes back to the queue
ch.basic_nack(delivery_tag=tag, requeue=True)

# Discard: message is removed (or dead-lettered if configured)
ch.basic_nack(delivery_tag=tag, requeue=False)

Requeueing without caution creates infinite loops. Use a redelivery counter or dead letter exchange.

Multiple Acknowledgements

Acknowledge multiple messages at once for batch processing:

# Ack a single message
ch.basic_ack(delivery_tag=tag)

# Ack this message and all earlier unacked messages
ch.basic_ack(delivery_tag=tag, multiple=True)

Use multiple acks carefully. They acknowledge all preceding unacked messages, not just the current one.

Common Mistakes

1. Using Auto-Ack in Production

Auto-ack removes messages on delivery, not on processing success. If the consumer crashes mid-processing, the message is lost permanently.

2. Not Acknowledging at All

If you never ack or nack, unacked message count grows until the consumer runs out of memory or hits the channel limit. Messages are held in memory on the broker.

3. Infinite Requeue Loops

Nacking with requeue=true without tracking delivery count creates infinite loops. The message is delivered, rejected, requeued, and delivered again forever.

4. Acknowledging Before Processing

Acking before the work is done defeats the purpose. If the consumer crashes after the ack, the message is considered processed and is removed.

5. Forgetting the Delivery Tag

The delivery tag identifies which message to ack. Always use method.delivery_tag from the callback. Hardcoding tags causes incorrect acknowledgements.

Practice Questions

1. What is the difference between auto-ack and manual ack?

Auto-ack removes the message when delivered. Manual ack removes it only after the consumer explicitly acknowledges processing.

2. What happens if a consumer disconnects without acking?

The broker requeues the message for delivery to another consumer. This provides at-least-once delivery.

3. When should you use basic.nack?

When processing fails and you want to either requeue the message (requeue=true) or discard/dead-letter it (requeue=false).

4. What is the multiple flag in basic.ack?

It acknowledges the specified message and all earlier unacknowledged messages on the channel.

Challenge

Build a consumer that processes messages with retry logic: Process normally, nack with requeue on transient failures (max 3 redeliveries), then nack with requeue=false to dead letter after exhausting retries.

FAQ

Mini Project: Safe Consumer

import pika
import time
import json

class SafeConsumer:
    def __init__(self):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.ch = self.conn.channel()
        self.ch.queue_declare(queue='safe_queue', durable=True)
        self.ch.basic_qos(prefetch_count=1)
        self.redeliveries = {}

    def process(self, ch, method, properties, body):
        msg_id = properties.message_id or str(time.time())
        redelivery_count = self.redeliveries.get(msg_id, 0)

        try:
            print(f"Processing: {body.decode()}")
            if redelivery_count > 0:
                print(f"Redelivery attempt {redelivery_count}")
            time.sleep(0.1)
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print("Acknowledged")
            if msg_id in self.redeliveries:
                del self.redeliveries[msg_id]
        except Exception as e:
            print(f"Error: {e}")
            if redelivery_count < 3:
                self.redeliveries[msg_id] = redelivery_count + 1
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
                print(f"Requeued (attempt {redelivery_count + 1})")
            else:
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
                print("Dead-lettered after 3 attempts")

    def run(self):
        self.ch.basic_consume(queue='safe_queue', on_message_callback=self.process)
        print("Consumer ready")
        self.ch.start_consuming()

consumer = SafeConsumer()
consumer.run()

Expected output:

Consumer ready
Processing: some message
Acknowledged

What's Next

Now that you understand consumer acknowledgements, explore QoS prefetch for controlling message flow to consumers, then learn about RabbitMQ clustering for high availability.