RabbitMQ Queue Message Loss: Acknowledgment and Persistence Pitfalls

Introduction

Message queues are supposed to guarantee delivery. That is the whole point — you put a message in, and it comes out the other side reliably. So when I discovered that RabbitMQ was silently dropping messages in my trading signal pipeline, it was deeply concerning. Messages were being published to the queue, but some consumers were never receiving them. There were no errors in the logs, no connection drops, just missing messages.

The root cause turned out to be a combination of three issues that individually seemed harmless but together created a perfect storm for message loss: auto-ack mode, non-durable queues, and an aggressive prefetch count. Here is how I traced and fixed each one.

Environment

• RabbitMQ version: 3.12.3 (Erlang/OTP 26)
• Node: Single node on Ubuntu 22.04
• Client library: pika 1.3.2 (Python)
• Queue type: Classic queue (not quorum)
• Message type: Trading signals (~1KB each)
• Publisher rate: ~1,000 messages/second
• Consumer processes: 4 Python workers

Problem

I was publishing trading signals to a RabbitMQ queue and consuming them with 4 worker processes. The system should have processed every signal exactly once. Instead, I noticed gaps in the output — signals that were published but never appeared in the consumer logs.

I added logging to the publisher:

import pika

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

channel.queue_declare(queue='trading_signals')

published_count = 0
for signal in generate_signals():
    channel.basic_publish(
        exchange='',
        routing_key='trading_signals',
        body=json.dumps(signal)
    )
    published_count += 1

print(f"Published {published_count} signals")
connection.close()

The publisher reported publishing 10,000 signals. But the consumers only processed about 9,700 of them. A 3% message loss rate was unacceptable for a trading system.

Analysis

I started by checking the queue state:

rabbitmqctl list_queues name messages consumers memory
# Listing queues for vhost /
# name                messages  consumers  memory
# trading_signals     300       4          1048576

There were 300 messages sitting in the queue — those were messages that had been published but not yet consumed. This was normal for a steady-state queue. But the total messages published (10,000) minus messages consumed (9,700) should have left 300 in the queue, which matched. The issue was not that messages were stuck — it was that some messages had been consumed but not acknowledged.

I checked the consumer configuration:

def callback(ch, method, properties, body):
    signal = json.loads(body)
    process_signal(signal)
    # No ch.basic_ack() call!

channel.basic_qos(prefetch_count=100)
channel.basic_consume(
    queue='trading_signals',
    on_message_callback=callback
)

The consumer was missing ch.basic_ack(method.delivery_tag). Without acknowledgment, RabbitMQ assumes the consumer has successfully processed the message and removes it from the queue after the consumer disconnects. If the consumer crashes or disconnects before finishing processing, the message is lost.

But wait — I was not seeing consumer crashes. The messages were being processed but never acknowledged. So why were they disappearing?

The second issue was the queue durability. I checked the queue declaration:

channel.queue_declare(queue='trading_signals')

This was a non-durable queue. When RabbitMQ restarts or the queue is idle, non-durable queues can lose messages. But I had not restarted RabbitMQ recently, so this was not the immediate cause.

The third issue was the prefetch count. I had set prefetch_count=100, which meant RabbitMQ would send up to 100 unacknowledged messages to each consumer. With 4 consumers and no acknowledgments, RabbitMQ would send 400 messages and then stop delivering more. The 300 messages sitting in the queue were exactly those — messages that had been delivered but not acknowledged.

The real problem was that some consumers were processing messages slowly, and RabbitMQ's delivery timeout was causing messages to be requeued and delivered to other consumers, leading to duplicates and potential loss during requeue operations.

Solution

I made three critical changes to fix the message loss:

1. Add proper acknowledgment:

def callback(ch, method, properties, body):
    try:
        signal = json.loads(body)
        process_signal(signal)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        logger.error(f"Failed to process signal: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

channel.basic_consume(
    queue='trading_signals',
    on_message_callback=callback
)

2. Make the queue durable:

# Publisher
channel.queue_declare(
    queue='trading_signals',
    durable=True,
    arguments={
        'x-queue-type': 'classic',
        'x-message-ttl': 3600000,  # 1 hour TTL
        'x-dead-letter-exchange': 'dead_letter',
        'x-dead-letter-routing-key': 'trading_signals.dlq'
    }
)

# Consumer must declare with same parameters
channel.queue_declare(
    queue='trading_signals',
    durable=True,
    arguments={
        'x-queue-type': 'classic',
        'x-message-ttl': 3600000,
        'x-dead-letter-exchange': 'dead_letter',
        'x-dead-letter-routing-key': 'trading_signals.dlq'
    }
)

3. Tune prefetch and add connection recovery:

import pika
from pika.exceptions import ConnectionClosedByBroker

def create_connection():
    params = pika.ConnectionParameters(
        host='localhost',
        heartbeat=600,
        blocked_connection_timeout=300,
        connection_attempts=3,
        retry_delay=5
    )
    return pika.BlockingConnection(params)

connection = create_connection()
channel = connection.channel()

# Lower prefetch for more reliable processing
channel.basic_qos(prefetch_count=10)

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

I also created a dead letter queue to capture messages that fail processing:

# Set up dead letter exchange
channel.exchange_declare(exchange='dead_letter', exchange_type='direct', durable=True)
channel.queue_declare(queue='trading_signals.dlq', durable=True)
channel.queue_bind(
    queue='trading_signals.dlq',
    exchange='dead_letter',
    routing_key='trading_signals.dlq'
)

After deploying the fixes, I ran the same test:

# Publisher: 10,000 signals
# Consumer output: 10,000 signals processed
# Messages in queue: 0
# Dead letter queue: 0
# Message loss: 0

Zero message loss. I also added monitoring to track queue health:

#!/bin/bash
# monitor-rabbitmq.sh

QUEUE="trading_signals"
THRESHOLD=1000

while true; do
    MSG_COUNT=$(rabbitmqctl list_queues name messages -q | grep "$QUEUE" | awk '{print $2}')

    if [ "$MSG_COUNT" -gt "$THRESHOLD" ]; then
        echo "$(date): WARNING: Queue $QUEUE has $MSG_COUNT messages (threshold: $THRESHOLD)"
    fi

    sleep 10
done

Lessons Learned

1. Always acknowledge messages — Without basic_ack(), messages are silently removed from the queue after the consumer disconnects. This is the most common cause of message loss in RabbitMQ.
2. Use durable queues — Non-durable queues lose all their messages if RabbitMQ restarts. Always declare queues as durable=True.
3. Set up dead letter queues — Messages that fail processing should go to a DLQ instead of being lost. This gives you a chance to inspect and retry failed messages.
4. Tune prefetch count — A high prefetch count with slow consumers can lead to message accumulation in the consumer's memory. Lower prefetch values ensure more even distribution.
5. Monitor queue depth — A growing queue depth is an early warning sign of consumer problems. Set up alerts for queue depth thresholds.

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.