QoS Prefetch in RabbitMQ — Complete Guide
In this tutorial, you will learn about QoS Prefetch in RabbitMQ. We cover key concepts, practical examples, and best practices to help you master this topic.
QoS prefetch limits how many unacknowledged messages a consumer can hold, enabling fair load distribution and preventing consumer overwhelm.
What You Learn
You will learn how QoS prefetch works, how to set prefetch count and size, and how it affects consumer throughput and fairness in RabbitMQ.
Why It Matters
Without QoS prefetch, RabbitMQ delivers all available messages to a consumer as fast as it can. Fast consumers get flooded with messages while slow consumers starve. Prefetch ensures fair distribution and prevents memory exhaustion.
Real-World Use
DodaZIP uses QoS prefetch with prefetch_count=1 for archive extraction workers. Each worker processes one archive at a time. If a worker crashes, only one message needs redelivery. This minimizes reprocessing work.
How QoS Prefetch Works
flowchart LR
B[Broker] -->|deliver 1 msg| C1[Consumer 1 prefetch=1]
B -->|deliver 1 msg| C2[Consumer 2 prefetch=1]
B -->|deliver 1 msg| C3[Consumer 3 prefetch=1]
C1 -->|ack| B
B -->|deliver next| C1
style B fill:#f90,color:#fff
With prefetch=1, each consumer holds at most one unacked message. Only after acking does it receive the next message.
Basic Prefetch Configuration
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='prefetch_demo', durable=True)
ch.basic_qos(prefetch_count=1)
print("QoS set: prefetch_count=1")
for i in range(5):
ch.basic_publish(exchange='', routing_key='prefetch_demo', body=f'msg_{i}')
def callback(ch, method, properties, body):
import time
print(f"Received: {body.decode()}")
time.sleep(0.5)
ch.basic_ack(delivery_tag=method.delivery_tag)
print(f"Acked: {body.decode()}")
ch.basic_consume(queue='prefetch_demo', on_message_callback=callback, auto_ack=False)
ch.start_consuming()
Expected output:
QoS set: prefetch_count=1
Received: msg_0
Acked: msg_0
Received: msg_1
Acked: msg_1
Prefetch Size
Limit total unacked message body size:
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='size_demo', durable=True)
ch.basic_qos(prefetch_count=10, prefetch_size=65536)
print("QoS: max 10 messages or 64KB total")
ch.basic_publish(exchange='', routing_key='size_demo', body='small_payload')
def cb(ch, method, properties, body):
print(f"Received ({len(body)} bytes): {body.decode()}")
ch.basic_ack(delivery_tag=method.delivery_tag)
ch.basic_consume(queue='size_demo', on_message_callback=cb)
ch.start_consuming()
Expected output:
QoS: max 10 messages or 64KB total
Received (12 bytes): small_payload
Prefetch with Multiple Consumers
import pika
import threading
import time
def start_consumer(name, prefetch_count):
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='multi_prefetch', durable=True)
ch.basic_qos(prefetch_count=prefetch_count)
def cb(ch, method, properties, body):
print(f"[{name}] Processing: {body.decode()}")
time.sleep(0.5)
ch.basic_ack(delivery_tag=method.delivery_tag)
print(f"[{name}] Done: {body.decode()}")
ch.basic_consume(queue='multi_prefetch', on_message_callback=cb)
ch.start_consuming()
threads = []
for i in range(3):
t = threading.Thread(target=start_consumer, args=(f'C{i}', 1), daemon=True)
t.start()
threads.append(t)
time.sleep(0.5)
pub = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
pub_ch = pub.channel()
pub_ch.queue_declare(queue='multi_prefetch', durable=True)
for i in range(6):
pub_ch.basic_publish(exchange='', routing_key='multi_prefetch', body=f'task_{i}')
time.sleep(3)
# Output will show round-robin distribution
Expected output:
[C0] Processing: task_0
[C1] Processing: task_1
[C2] Processing: task_2
[C0] Done: task_0
[C1] Done: task_1
[C2] Done: task_2
[C0] Processing: task_3
Global vs Per-Consumer Prefetch
By default, prefetch applies per consumer. Set global_qos=True to apply it across all consumers on the channel:
# Per-consumer (default): each consumer gets prefetch_count unacked
ch.basic_qos(prefetch_count=5)
# Global: all consumers share the prefetch limit
ch.basic_qos(prefetch_count=10, global_qos=True)
Choosing the Right Prefetch Value
| Workload | Prefetch Count | Reason |
|---|---|---|
| CPU-bound tasks | 1 | One task at a time per worker |
| I/O-bound tasks | 10-100 | Workers wait on I/O, more inflight = better utilization |
| Memory-sensitive | Low | Limit unacked messages to control memory usage |
| Batch processing | 100+ | Process in batches for throughput |
Common Mistakes
1. Setting Prefetch Too High
High prefetch floods consumers with messages. A slow consumer accumulates unacked messages in memory, risking OOM crashes.
2. Setting Prefetch Too Low
Prefetch=1 on I/O-bound tasks wastes throughput while workers wait for I/O to complete. Increase prefetch for non-blocking workloads.
3. Not Setting Prefetch at All
Without prefetch, RabbitMQ delivers messages as fast as possible. One consumer gets all messages while others get none.
4. Confusing Prefetch with Queue Length
Prefetch limits inflight messages per consumer. Queue length is the total unprocessed messages. They are independent settings.
5. Not Calling basic_qos Before Consuming
Set QoS prefetch before calling basic_consume. Setting it after may not apply to already registered consumers.
Practice Questions
1. What does prefetch_count control?
The maximum number of unacknowledged messages delivered to a consumer at one time.
2. Why would you set prefetch_count=1?
To ensure at-most-one message per consumer. Each message is fully processed before the next is delivered. Ideal for CPU-bound tasks.
3. What is the difference between per-consumer and global prefetch?
Per-consumer limits each consumer individually. Global limits all consumers on the channel combined.
4. What happens if prefetch_count is 0?
No limit. The broker delivers messages as fast as possible, which can overwhelm consumers.
Challenge
Design a prefetch Strategy for a video processing system: 3 workers, each transcoding takes 10-30 seconds, workers have 512MB RAM each, messages contain 100KB video references. Calculate appropriate prefetch count and size.
FAQ
Mini Project: Load-Balanced Worker Pool
import pika
import threading
import time
import random
class Worker:
def __init__(self, name, prefetch):
self.name = name
self.prefetch = prefetch
self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
self.ch = self.conn.channel()
self.ch.queue_declare(queue='worker_queue', durable=True)
self.ch.basic_qos(prefetch_count=prefetch)
self.processed = 0
def process(self, ch, method, properties, body):
work_time = random.uniform(0.1, 1.0)
print(f"[{self.name}] Work on {body.decode()} for {work_time:.1f}s")
time.sleep(work_time)
ch.basic_ack(delivery_tag=method.delivery_tag)
self.processed += 1
print(f"[{self.name}] Done ({self.processed} total)")
def run(self):
self.ch.basic_consume(queue='worker_queue', on_message_callback=self.process)
self.ch.start_consuming()
workers = [Worker('W1', 1), Worker('W2', 2), Worker('W3', 3)]
t1 = threading.Thread(target=workers[0].run, daemon=True)
t2 = threading.Thread(target=workers[1].run, daemon=True)
t3 = threading.Thread(target=workers[2].run, daemon=True)
t1.start(); t2.start(); t3.start()
pub = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
pub_ch = pub.channel()
pub_ch.queue_declare(queue='worker_queue', durable=True)
for i in range(20):
pub_ch.basic_publish(exchange='', routing_key='worker_queue', body=f'task_{i}')
time.sleep(5)
What's Next
Now that you understand QoS prefetch, explore RabbitMQ clustering for scaling across multiple nodes, then learn about queue mirroring for high availability.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro