Producer-Consumer Pattern — Complete Guide

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

The producer-consumer pattern separates message creation from message processing, enabling independent scaling of publishers and subscribers in distributed systems.

What You'll Learn

By the end of this lesson, you will understand the producer-consumer pattern, its benefits for decoupling and scalability, how to implement it in multiple languages, and best practices for both sides.

Why It Matters

Tightly coupled systems break when one component fails. If your web server directly calls an image processor and the image processor is down, the web server returns errors. The producer-consumer pattern inserts a buffer between them so each side operates independently.

Real-World Use

A file upload service accepts files and publishes messages to a queue. A separate worker fleet pulls messages and processes files. During peak hours, the upload service scales to handle more uploads. The worker fleet scales independently based on queue depth.

Producer-Consumer Flow

flowchart LR
    subgraph "Producers"
        P1[Web App]
        P2[API Service]
        P3[Cron Job]
    end
    subgraph "Broker"
        Q[Queue]
    end
    subgraph "Consumers"
        C1[Worker 1]
        C2[Worker 2]
        C3[Worker N]
    end
    P1 --> Q
    P2 --> Q
    P3 --> Q
    Q --> C1
    Q --> C2
    Q --> C3
    style Q fill:#f90,color:#fff

Producers create messages and send them to the broker. Consumers pick up messages from the broker and Process them. Neither side knows about the other.

Producer Responsibilities

The producer creates messages with enough context for the consumer to do its work. A good message contains the event type, a unique ID, a timestamp, and the data payload. The producer does not wait for processing to complete.

import pika
import json
import uuid

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='orders', durable=True)

order = {
    'order_id': str(uuid.uuid4()),
    'user_id': 42,
    'items': ['product_1', 'product_2'],
    'total': 59.99,
    'timestamp': '2026-06-28T10:00:00Z'
}

channel.basic_publish(
    exchange='',
    routing_key='orders',
    body=json.dumps(order),
    properties=pika.BasicProperties(
        delivery_mode=2,
        content_type='application/json',
        message_id=order['order_id']
    )
)
print(f"Published order {order['order_id']}")
connection.close()

Expected output:

Published order 550e8400-e29b-41d4-a716-446655440000

Consumer Responsibilities

The consumer pulls messages from the queue, processes them, and acknowledges completion. If processing fails, the consumer can reject or retry the message.

import pika
import json

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

def process_order(ch, method, properties, body):
    order = json.loads(body)
    print(f"Processing order {order['order_id']}")
    print(f"  User: {order['user_id']}")
    print(f"  Items: {', '.join(order['items'])}")
    print(f"  Total: ${order['total']}")
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(f"Order {order['order_id']} completed")

channel.basic_consume(queue='orders', on_message_callback=process_order)
print("Waiting for orders...")
channel.start_consuming()

Expected output:

Waiting for orders...
Processing order 550e8400-e29b-41d4-a716-446655440000
  User: 42
  Items: product_1, product_2
  Total: $59.99
Order 550e8400-e29b-41d4-a716-446655440000 completed

Scaling Producers and Consumers Independently

Think of the queue as a buffer between two factories. The producing Factory works at its own speed, placing finished goods on a conveyor belt. The consuming factory picks goods from the belt at its own speed. If the consuming factory slows down, the belt fills up. If it speeds up, the belt empties.

This independence is powerful. You can add more producers without affecting consumers, and vice versa. Each side scales to meet its own demand.

# Simulating scale-out: start multiple consumers
import threading
import time

def start_consumer(name):
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.queue_declare(queue='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='tasks', on_message_callback=cb)
    ch.start_consuming()

for i in range(4):
    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='tasks', durable=True)

for i in range(20):
    ch.basic_publish(
        exchange='', routing_key='tasks',
        body=f'task-{i}',
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"Published task-{i}")

conn.close()
time.sleep(3)

Expected output:

Published task-0
[Worker-0] Processing: task-0
Published task-1
[Worker-1] Processing: task-1
... (tasks distributed across 4 workers)

Common Mistakes

1. Making the Producer Wait for the Consumer

The producer should never block waiting for the consumer to finish. That defeats the purpose of async messaging. If you need a response, use the request-reply pattern instead.

2. Publishing Messages Without Unique IDs

Without unique IDs, consumers cannot deduplicate messages. If the same message is delivered twice, it is processed twice. Always include a unique message ID.

3. Putting Business Logic in the Producer

The producer should only format and send messages. Business logic belongs in consumers. This keeps the producer fast and the system flexible.

4. Consumers That Never Acknowledge

If a consumer forgets to acknowledge, messages stay in the unacknowledged state. Eventually the queue fills up and new messages are rejected. Always acknowledge after successful processing.

5. Overloading the Message with Context

A message should contain enough data for processing but not more. Sending a full database row as a message wastes bandwidth and couples the consumer to the producer's schema. Send only what the consumer needs.

Practice Questions

1. What is the main benefit of separating producers and consumers?

Independent scaling. Producers can scale to handle more input without affecting consumers. Consumers can scale independently based on processing capacity. One side can fail without affecting the other.

2. How does a consumer know a message is available?

The consumer polls the broker or the broker pushes messages to the consumer. In RabbitMQ, the broker pushes messages to consumers subscribed to a queue. In Kafka, consumers poll for new messages.

3. What happens if a consumer crashes mid-processing?

With manual acknowledgments, the message remains unacknowledged. When the broker detects the consumer is gone, it re-queues the message for another consumer. The message is not lost.

4. Can a single consumer handle multiple message types?

Yes. The message can include a type field. The consumer inspects the type and routes to the appropriate handler function. This is common in event-driven systems.

Challenge

Design an order processing system with three producer types (web, mobile, API) and four consumer types (payment, inventory, email, shipping). Each consumer type may have multiple instances. Ensure order processing is fault-tolerant and scalable.

FAQ

Mini Project: Multi-Language Producer-Consumer

import pika
import json
import time

class MessageProducer:
    def __init__(self):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.ch = self.conn.channel()
        self.ch.queue_declare(queue='tasks', durable=True)

    def publish(self, task_type, data):
        msg = {'type': task_type, 'data': data, 'timestamp': time.time()}
        self.ch.basic_publish(
            exchange='', routing_key='tasks',
            body=json.dumps(msg),
            properties=pika.BasicProperties(delivery_mode=2)
        )
        print(f"Published {task_type}")
        return msg

    def close(self):
        self.conn.close()

producer = MessageProducer()
for i in range(5):
    producer.publish('email', {'to': f'user{i}@example.com', 'subject': f'Hello {i}'})
producer.close()

Expected output:

Published email
Published email
Published email
Published email
Published email

What's Next

Now that you understand the producer-consumer pattern, explore message formats to choose between JSON, Avro, and Protobuf, then learn about message persistence for guaranteed delivery.