RabbitMQ’s acknowledgment system is designed to ensure that messages are processed reliably, meaning no message is lost and no message is processed more than once.
Here’s a look at how it works with a consumer:
Imagine you have a simple producer sending messages to a queue named task_queue. A consumer then connects to this queue to process messages.
# producer.py
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
))
print(f" [x] Sent '{message}'")
connection.close()
# consumer.py
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
def callback(ch, method, properties, body):
print(f" [x] Received {body.decode()}")
# Simulate work
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag) # Acknowledge receipt and processing
channel.basic_qos(prefetch_count=1) # Only one unacknowledged message at a time
channel.basic_consume(queue='task_queue',
auto_ack=False, # We will manually acknowledge
on_message_callback=callback)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
When the consumer receives a message, the callback function is executed. Inside this callback, after simulating some work (by sleeping for a duration based on the number of dots in the message), the ch.basic_ack(delivery_tag=method.delivery_tag) line is crucial. This tells RabbitMQ that the message has been successfully processed and can be removed from the queue.
If the consumer crashes before basic_ack is called, RabbitMQ will, upon realizing the consumer is no longer connected, re-queue the unacknowledged message and deliver it to another (or the same, if it restarts) connected consumer. This is the core of reliable delivery.
The channel.basic_qos(prefetch_count=1) line is also important. It tells RabbitMQ to send only one message to the consumer at a time. The consumer will only receive a new message after it has acknowledged the previous one. This prevents a fast consumer from overwhelming a slow one, or a consumer from receiving multiple messages and then crashing before acknowledging any of them, leading to a backlog of unacknowledged messages.
The mental model here is that RabbitMQ acts as a reliable post office. It holds onto letters (messages) until it gets confirmation from the recipient (consumer) that the letter has been received and understood. If the recipient disappears before confirming, the post office assumes the letter was lost and tries to deliver it again.
The auto_ack=False in basic_consume is the switch that enables this manual acknowledgment mechanism. If auto_ack were True, RabbitMQ would consider a message acknowledged as soon as it’s delivered, and if the consumer crashed immediately after delivery, the message would be lost.
The one thing most people don’t realize is that basic_ack acknowledges all messages up to and including the delivery_tag specified. If you’ve received messages with delivery tags 1, 2, and 3, and you call basic_ack with delivery tag 3, messages 1 and 2 are also implicitly acknowledged. This is important for performance as it reduces the number of network round trips.
The next step in building robust consumers involves understanding basic_nack and basic_reject for handling messages that cannot be processed.