TopolearnAcks decide whether a crash loses messages or redelivers them; getting them wrong causes data loss or infinite loops.
With manual acknowledgements (auto_ack=False, the default in pika's basic_consume), the broker keeps every delivered message as unacked until the consumer responds:
basic_ack(delivery_tag): done, delete the message.multiple=Trueacknowledges all up to that tag.basic_nack(delivery_tag, requeue=True): failed, put it back in the queue (at or near its original position) for redelivery.basic_nack(..., requeue=False)orbasic_reject: failed, discard it, or dead-letter it if the queue has a dead-letter exchange.
If the channel or connection closes with unacked messages, they are requeued automatically and redelivered with the redelivered flag set. This gives at-least-once processing: a message is never lost because a consumer crashed, but it may be processed twice.
Rules for robust consumers:
- Ack after the work and its side effects are committed, never before.
- Don't requeue forever on errors that will never succeed (malformed data): a poison message with
requeue=Trueloops endlessly. Reject without requeue and dead-letter it. Quorum queues enforce a delivery limit for this. - Ack on the same channel the message arrived on; a delivery tag is per channel.
- RabbitMQ closes channels that keep a message unacked longer than the consumer timeout (30 minutes by default). Very long tasks need a higher timeout or a different design.
Example
import json, pika
ch = pika.BlockingConnection(pika.ConnectionParameters("localhost")).channel()
ch.queue_declare(queue="emails", durable=True)
ch.basic_qos(prefetch_count=10)
def send_email(msg): print("sending to", msg["to"])
def on_message(channel, method, props, body):
try:
msg = json.loads(body)
except ValueError:
channel.basic_reject(method.delivery_tag, requeue=False) # poison: drop/DLX
return
try:
send_email(msg)
channel.basic_ack(method.delivery_tag) # only after success
except ConnectionError:
channel.basic_nack(method.delivery_tag, requeue=True) # transient: retry
if method.redelivered:
print("this message was delivered before")
ch.basic_consume(queue="emails", on_message_callback=on_message) # manual acks
ch.start_consuming()Remember
- Ack after the work is done; unacked messages are redelivered if the channel dies
- nack/reject with requeue=False drops or dead-letters poison messages
- Manual acks give at-least-once delivery, so duplicates are possible
Watch out for
- Requeueing a message that always fails, creating a hot redelivery loop.