Fix RabbitMQ Queue Messages Not Acknowledged

What's Actually Happening

RabbitMQ messages remain in unacknowledged state after being delivered to consumers. The messages block the queue and may cause memory pressure.

The Error You'll See

Unacked messages:

```bash $ rabbitmqctl list_queues name messages_unacknowledged

myqueue 500 # 500 unacknowledged messages ```

Queue stuck:

```bash $ rabbitmqctl list_queues name messages consumers

myqueue 500 1 # Messages piling up with only 1 consumer ```

Management UI shows:

Ready: 0
Unacked: 500
Total: 500
# All messages unacked, none ready for delivery

Why This Happens

1. 1.Consumer not acking - Consumer fails to send acknowledgment
2. 2.Consumer crash - Consumer dies without acking
3. 3.Prefetch too high - Many messages delivered but not processed
4. 4.Processing errors - Exception prevents ack
5. 5.Manual ack missing - Auto-ack disabled but no manual ack
6. 6.Network issues - Ack lost in transit

Step 1: Check Queue Status

```bash # List queues with unacked messages: rabbitmqctl list_queues name messages messages_unacknowledged consumers

# Output: # name messages unacked consumers # myqueue 0 500 1

# Check via management API: curl -u guest:guest http://localhost:15672/api/queues/%2F/myqueue

# Detailed queue info: rabbitmqctl list_queues name messages_unacknowledged consumer_utilisation

# consumer_utilisation: fraction of time consumers are active

# Check all queues: rabbitmqctl list_queues name messages messages_unacknowledged

# Check queue with arguments: rabbitmqctl list_queues name arguments ```

Step 2: Check Consumer Connections

```bash # List connections: rabbitmqctl list_connections name state channels

# List channels: rabbitmqctl list_channels consumer_count messages_unacknowledged

# Find consumers on queue: rabbitmqctl list_consumers

# Output shows: # queue_name channel_pid consumer_tag ack_required prefetch_count

# Check consumer process: ps aux | grep your-consumer

# If consumer down: # Restart consumer application systemctl restart my-consumer

# Check consumer logs: journalctl -u my-consumer -n 50 ```

Step 3: Check Ack Handling

```python # Correct ack handling (Python/pika):

# Wrong: Missing ack def on_message(channel, method, properties, body): process_message(body) # Missing: channel.basic_ack(delivery_tag=method.delivery_tag)

# Correct: Ack after processing def on_message(channel, method, properties, body): try: process_message(body) channel.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: # Nack with requeue channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

# Consumer setup: channel.basic_consume( queue='myqueue', on_message_callback=on_message, auto_ack=False # Manual ack required ) ```

```java // Java/RabbitMQ client:

// Wrong: Missing ack deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); processMessage(message); // Missing: channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); };

// Correct: deliverCallback = (consumerTag, delivery) -> { try { String message = new String(delivery.getBody(), "UTF-8"); processMessage(message); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (Exception e) { channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); } }; ```

Step 4: Adjust Prefetch Count

```bash # Check current prefetch: rabbitmqctl list_consumers | grep myqueue

# Prefetch too high causes many unacked messages: # Consumer receives prefetch_count messages at once

# Set appropriate prefetch: # Python: channel.basic_qos(prefetch_count=10)

# Java: channel.basicQos(10);

# Rule: prefetch_count should match processing capacity # Lower prefetch = less unacked messages # But also = lower throughput

# Recommended: Start with prefetch=1, increase if needed channel.basic_qos(prefetch_count=1)

# Global prefetch (shared across consumers): channel.basic_qos(prefetch_count=10, global_qos=True)

# Restart consumer after changing prefetch ```

Step 5: Handle Consumer Failures

```bash # If consumer crashed with unacked messages:

# 1. Messages will be requeued when connection closes # 2. But if connection still open, messages stay unacked

# Force close consumer connection: rabbitmqctl close_connection <connection_pid> "Consumer restart"

# Or via management API: curl -u guest:guest -X DELETE http://localhost:15672/api/connections/<connection_name>

# This will requeue unacked messages

# Check connections: rabbitmqctl list_connections pid name

# Close specific connection: rabbitmqctl close_connection <pid> "Manual close"

# Messages become ready again: rabbitmqctl list_queues name messages messages_unacknowledged ```

Step 6: Check for Dead Consumers

```bash # List consumers and their channels: rabbitmqctl list_consumers

# Check channel state: rabbitmqctl list_channels pid state messages_unacknowledged

# Dead consumer: connection open but not processing

# Set consumer timeout: # In rabbitmq.conf: consumer_timeout = 1800000 # 30 minutes

# After timeout, unacked messages requeued

# Or use heartbeat: # In consumer config: # heartbeat = 60 # seconds

# Check heartbeat settings: rabbitmqctl eval 'application:get_env(rabbit, heartbeat).'

# List stale connections: rabbitmqctl list_connections name state last_received_oct ```

Step 7: Implement Dead Letter Queue

```bash # Create DLQ: rabbitmqadmin declare queue name=myqueue.dlq durable=true

# Configure queue with DLQ: rabbitmqadmin declare queue name=myqueue \ durable=true \ arguments='{"x-dead-letter-exchange":"", "x-dead-letter-routing-key":"myqueue.dlq"}'

# Consumer with DLQ handling: # After max retries, reject without requeue channel.basic_reject(delivery_tag=tag, requeue=False) # Message goes to DLQ

# Check DLQ: rabbitmqctl list_queues name messages

# Or with max length and overflow: rabbitmqadmin declare queue name=myqueue \ arguments='{"x-max-length":10000, "overflow":"reject-publish-dlx", "x-dead-letter-exchange":"my.dlx"}' ```

Step 7: Monitor Ack Metrics

```bash # RabbitMQ management metrics: curl -u guest:guest http://localhost:15672/api/queues/%2F/myqueue | jq '.messages_unacknowledged'

# Prometheus metrics (if enabled): curl http://localhost:15692/metrics | grep rabbitmq_queue_messages_unacked

# Key metrics: # rabbitmq_queue_messages - total messages # rabbitmq_queue_messages_unacked - unacknowledged # rabbitmq_queue_consumers - consumer count

# Create monitoring script: cat << 'EOF' > /usr/local/bin/monitor-rabbitmq.sh #!/bin/bash

echo "=== Queues with Unacked Messages ===" rabbitmqctl list_queues name messages messages_unacknowledged | awk '$3 > 0'

echo "" echo "=== Consumer Status ===" rabbitmqctl list_consumers | grep -c myqueue

echo "" echo "=== Connection Status ===" rabbitmqctl list_connections name state

echo "" echo "=== Memory Usage ===" rabbitmqctl status | grep -A 5 "Memory" EOF

chmod +x /usr/local/bin/monitor-rabbitmq.sh

# Alert for high unacked: - alert: RabbitMQUnackedMessages expr: rabbitmq_queue_messages_unacked > 1000 for: 5m labels: severity: warning annotations: summary: "RabbitMQ queue has high unacked messages" ```

Step 8: Check Flow Control

```bash # If RabbitMQ in flow control, acks may be delayed:

# Check flow control status: rabbitmqctl list_connections name flow_control

# Flow control activated when: # - Memory alarm triggered # - Disk alarm triggered # - Socket descriptors exhausted

# Check alarms: rabbitmqctl status | grep -A 10 "Alarms"

# Clear memory alarm (if safe): rabbitmqctl clear_memory_alarm

# Check memory watermark: rabbitmqctl eval 'application:get_env(rabbit, vm_memory_high_watermark).'

# Default: 0.4 (40% of RAM)

# Increase watermark: # In rabbitmq.conf: vm_memory_high_watermark.relative = 0.6

# Restart RabbitMQ: systemctl restart rabbitmq-server ```

Step 9: Use Nack Appropriately

```python # Nack with requeue (for retry): channel.basic_nack(delivery_tag=tag, requeue=True)

# Nack without requeue (discard/DLQ): channel.basic_nack(delivery_tag=tag, requeue=False)

# Reject single message: channel.basic_reject(delivery_tag=tag, requeue=True)

# Multiple messages (up to delivery_tag): channel.basic_nack(delivery_tag=tag, multiple=True, requeue=True)

# Common pattern: Retry with backoff def on_message(channel, method, properties, body): retry_count = properties.headers.get('x-retry-count', 0)

if retry_count > 3: # Send to DLQ channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False) return

try: process_message(body) channel.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: # Increment retry count and requeue new_headers = {'x-retry-count': retry_count + 1} # Would need to republish with new headers channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True) ```

Step 10: Purge Unacked Messages

```bash # Emergency: Purge all messages (including unacked):

# Warning: This deletes all messages! rabbitmqctl purge_queue myqueue

# Or via API: curl -u guest:guest -X DELETE http://localhost:15672/api/queues/%2F/myqueue/contents

# To requeue unacked without purge: # Close consumer connection: rabbitmqctl close_connection <pid> "Requeue unacked"

# Check messages requeued: rabbitmqctl list_queues name messages messages_unacknowledged

# Now messages are ready for new consumers ```

RabbitMQ Unacked Messages Checklist

Verify the Fix

```bash # After fixing ack issues

# 1. Check unacked count rabbitmqctl list_queues name messages_unacknowledged // Low or zero

# 2. Send test message rabbitmqadmin publish routing_key=myqueue payload="test message" // Published

# 3. Check message processed rabbitmqctl list_queues name messages messages_unacknowledged // Message acked, queue empty

# 4. Monitor consumer logs journalctl -u my-consumer -f // Processing messages, acking

# 5. Test under load rabbitmqadmin publish routing_key=myqueue payload="load test" --rate=100 // Messages acked promptly

# 6. Check queue health /usr/local/bin/monitor-rabbitmq.sh // No issues ```

Related Issues

• [Fix RabbitMQ Queue Messages Not Being Consumed](/articles/fix-rabbitmq-queue-messages-not-being-consumed)
• [Fix RabbitMQ Memory High Watermark](/articles/fix-rabbitmq-memory-high-watermark)
• [Fix Kafka Consumer Lag Growing](/articles/fix-kafka-consumer-lag-growing)