Competing Consumers — Complete Guide

In this tutorial, you will learn about Competing Consumers. We cover key concepts, practical examples, and best practices to help you master this topic.

Competing consumers scale message processing by running multiple consumer instances that all receive messages from the same queue, each processing one message at a time.

What You'll Learn

By the end of this lesson, you will understand the competing consumers pattern, how to configure fair dispatch, how to scale consumers horizontally, and when to use this pattern for parallel task processing.

Why It Matters

A single consumer can Process only one message at a time. If each message takes 1 second, throughput is 1 msg/s. With 10 competing consumers, throughput increases to 10 msg/s. Competing consumers are the primary way to scale message processing.

Real-World Use

A video transcoding platform receives hundreds of uploads per minute. Each video takes 2-10 minutes to transcode. A single transcoder cannot keep up. The platform runs 50 consumer instances behind a queue. Each picks up a video, processes it, and picks up the next.

Competing Consumers Flow

flowchart LR
    P[Producer] --> Q[Work Queue]
    Q --> C1[Consumer 1]
    Q --> C2[Consumer 2]
    Q --> C3[Consumer 3]
    Q --> C4[Consumer N]
    C1 --> D1[Processes message A]
    C2 --> D2[Processes message B]
    C3 --> D3[Processes message C]
    C4 --> D4[Processes message D]
    style Q fill:#f90,color:#fff

All consumers subscribe to the same queue. The broker distributes messages across consumers. Each message is delivered to exactly one consumer.

Fair Dispatch

Without fair dispatch, RabbitMQ sends messages to consumers in round-robin order. If consumer A takes 10 seconds per message and consumer B takes 1 second, they each get the same number of messages. B finishes quickly and waits, while A falls behind.

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks', durable=True)
channel.basic_qos(prefetch_count=1)

def callback(ch, method, properties, body):
    task_time = int(body.decode().split(':')[1])
    print(f"Processing task (takes {task_time}s)")
    time.sleep(task_time)
    print("Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='tasks', on_message_callback=callback)
channel.start_consuming()

prefetch_count=1 tells RabbitMQ to send only one message at a time to each consumer. RabbitMQ waits for the ack before sending the next message. This ensures fast consumers get more work than slow ones.

# Producer with varied task times
import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='tasks', durable=True)

import random
for i in range(10):
    task_time = random.randint(1, 5)
    ch.basic_publish(
        exchange='', routing_key='tasks',
        body=f'task-{i}:{task_time}',
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"Queued task-{i} ({task_time}s)")

conn.close()

Expected output:

Queued task-0 (3s)
Queued task-1 (1s)
Queued task-2 (5s)
...

Scaling Consumers

import pika
import threading
import time

def start_consumer(name):
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.queue_declare(queue='scalable_tasks', durable=True)
    ch.basic_qos(prefetch_count=1)

    def cb(c, m, p, body):
        print(f"[{name}] Processing: {body.decode()}")
        time.sleep(0.5)
        c.basic_ack(delivery_tag=m.delivery_tag)

    ch.basic_consume(queue='scalable_tasks', on_message_callback=cb)
    print(f"[{name}] Started")
    ch.start_consuming()

for i in range(3):
    t = threading.Thread(target=start_consumer, args=(f'Worker-{i}',), daemon=True)
    t.start()

time.sleep(1)
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='scalable_tasks', durable=True)

for i in range(9):
    ch.basic_publish(exchange='', routing_key='scalable_tasks',
                     body=f'Task-{i}', properties=pika.BasicProperties(delivery_mode=2))

conn.close()
time.sleep(3)

Expected output:

[Worker-0] Started
[Worker-1] Started
[Worker-2] Started
[Worker-0] Processing: Task-0
[Worker-1] Processing: Task-1
[Worker-2] Processing: Task-2
[Worker-0] Processing: Task-3
...

Three workers process nine tasks in parallel. Total time is 1.5 seconds instead of 4.5 seconds with a single worker.

Dynamic Scaling

You can add or remove consumers at any time without stopping the system. When a new consumer starts, RabbitMQ begins sending it messages immediately. When a consumer stops, its unacknowledged messages are redistributed to remaining consumers.

Common Mistakes

1. Not Setting Prefetch Count

Without prefetch, one consumer can hoard all messages. The first consumer to connect gets all messages, and others starve. Always set prefetch_count=1 for fair distribution.

2. Running Competing Consumers with Ordered Messages

Competing consumers break message ordering. If message 2 arrives before message 1 is acknowledged, consumer B may process message 2 before consumer A finishes message 1. Do not use competing consumers when order matters.

3. Ignoring Slow Consumer Detection

One slow consumer can bottleneck the entire system if prefetch is misconfigured. Monitor individual consumer processing times and remove consumers that are consistently slower than others.

4. Using Too Many Consumers

More consumers does not always mean more throughput. If consumers share a bottleneck (database, API rate limit), adding more consumers increases contention. Measure system limits first.

5. Not Rebalancing on Scale-Down

When consumers shut down, their unacknowledged messages are re-queued. If many consumers shut down simultaneously, the queue may get a flood of re-queued messages. Implement graceful shutdown.

Practice Questions

1. How do competing consumers increase throughput?

Each consumer processes messages independently in parallel. With N consumers, theoretical throughput is N times that of a single consumer (limited by shared resources like database).

2. What is fair dispatch?

Fair dispatch ensures that faster consumers get more work. RabbitMQ sends each consumer only one message at a time (prefetch=1). When a consumer finishes and acks, it gets the next message.

3. What happens when a competing consumer crashes?

Its unacknowledged messages are re-queued by the broker. Other consumers pick them up. No messages are lost as long as the queue is durable and messages are persistent.

4. Can competing consumers guarantee ordering?

No. Different consumers process messages at different speeds. Message 2 may complete before message 1. Use a single consumer or partitioned ordering for sequential processing.

Challenge

Design a competing consumer system for an email sending service. 100,000 emails per hour need to be sent. Each email takes 0.5-2 seconds to send. Calculate the number of consumers needed and design for fair dispatch.

FAQ

Mini Project: Auto-Scaling Worker Pool

import pika
import threading
import time
import random

class AutoScaler:
    def __init__(self, min_workers=2, max_workers=10, target_qdepth=10):
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.target_qdepth = target_qdepth
        self.workers = {}
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.ch = self.conn.channel()
        self.ch.queue_declare(queue='autoscale_tasks', durable=True)

    def get_qdepth(self):
        q = self.ch.queue_declare(queue='autoscale_tasks', passive=True)
        return q.method.message_count

    def start_worker(self, wid):
        def worker():
            conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
            ch = conn.channel()
            ch.queue_declare(queue='autoscale_tasks', durable=True)
            ch.basic_qos(prefetch_count=1)

            def cb(c, m, p, body):
                time.sleep(random.uniform(0.1, 0.5))
                c.basic_ack(delivery_tag=m.delivery_tag)

            ch.basic_consume(queue='autoscale_tasks', on_message_callback=cb)
            ch.start_consuming()

        t = threading.Thread(target=worker, daemon=True)
        self.workers[wid] = t
        t.start()

    def scale(self):
        depth = self.get_qdepth()
        current = len(self.workers)
        target = max(self.min_workers, min(self.max_workers, depth // self.target_qdepth + 1))

        if target > current:
            for i in range(current, target):
                self.start_worker(i)
            print(f"Scaled up: {current} -> {target} (depth: {depth})")
        elif target < current:
            print(f"Would scale down: {current} -> {target} (depth: {depth})")

scaler = AutoScaler()
scaler.scale()
print(f"Workers: {len(scaler.workers)}")

Expected output:

Scaled up: 0 -> 2 (depth: 15)
Workers: 2

What's Next

Now that you understand competing consumers, explore fanout exchanges for broadcast messaging, then learn about topic exchanges for routed message delivery.