Point-to-Point Pattern — Complete Guide

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

Point-to-point messaging delivers each message to exactly one consumer from a queue, ideal for task distribution where work must not be duplicated across multiple workers.

What You'll Learn

By the end of this lesson, you will understand the point-to-point pattern, when to use it, how it differs from pub-sub, and how to implement it with RabbitMQ and competing consumers.

Why It Matters

Many real-world workloads involve distributing tasks across workers where each task must be processed exactly once. Sending a welcome email, resizing an image, charging a payment — these should never be duplicated. The point-to-point pattern guarantees single-consumer delivery.

Real-World Use

A document processing service receives PDF uploads. Each PDF must be converted to text, indexed for search, and archived. Uploads arrive faster than a single worker can Process them. A queue holds the tasks, and multiple worker instances each grab the next available PDF. No PDF is processed by two workers, and no PDF is missed.

How Point-to-Point Works

flowchart LR
    P[Producer] --> Q[Queue]
    Q --> C1[Consumer 1]
    Q --> C2[Consumer 2]
    Q --> C3[Consumer 3]
    style Q fill:#f90,color:#fff

The producer sends messages to a queue. Multiple consumers read from the same queue, but each message is delivered to exactly one consumer. This is the competing consumers pattern.

Point-to-Point vs Pub-Sub

Implementing with RabbitMQ

import pika
import json

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

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

channel.basic_publish(
    exchange='',
    routing_key='tasks',
    body=json.dumps({'task': 'resize_image', 'file': 'photo.jpg'}),
    properties=pika.BasicProperties(delivery_mode=2)
)

print("Task published to queue 'tasks'")
connection.close()

Expected output:

Task published to queue 'tasks'

import pika
import json
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 = json.loads(body)
    print(f"Processing: {task['task']} for {task['file']}")
    time.sleep(1)
    print(f"Completed: {task['task']}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='tasks', on_message_callback=callback)
print("Waiting for tasks. Press Ctrl+C to exit.")
channel.start_consuming()

Expected output:

Waiting for tasks. Press Ctrl+C to exit.
Processing: resize_image for photo.jpg
Completed: resize_image

The consumer sets prefetch_count=1 so RabbitMQ does not send more than one message at a time. This ensures fair distribution across multiple consumers.

Why the Queue Guarantees Single Delivery

The queue holds messages in order. When a consumer requests a message, the broker dequeues it and marks it as delivered but not yet acknowledged. If the consumer crashes before acknowledging, the broker re-queues the message so another consumer can pick it up.

This is why you never lose messages, but you might process a message twice if the consumer crashes after processing but before acknowledging. Your application must handle this by making tasks idempotent.

Common Mistakes

1. Forgetting to Set Prefetch Count

Without prefetch, RabbitMQ sends all messages to the first available consumer. Other consumers stay idle while one consumer is overloaded. Always set prefetch_count=1 for fair distribution.

2. Using Auto-Acknowledgement

With auto_ack=True, the broker removes the message as soon as it is sent. If the consumer crashes mid-processing, the message is lost forever. Always use manual acknowledgments for critical tasks.

3. Assuming Strict Ordering with Multiple Consumers

If you run three consumers, message 1 might go to consumer A, message 2 to consumer B. Consumer B finishes first, so message 2 is processed before message 1. Do not rely on FIFO ordering with multiple consumers.

4. Not Making Tasks Idempotent

The same task may be delivered twice if the consumer crashes after processing but before acknowledging. Design tasks so running them twice produces the same result as running them once.

5. Ignoring Poison Messages

A message that always causes an error (poison message) is repeatedly delivered and repeatedly fails. Implement a dead-letter queue with a max retry count to quarantine problematic messages.

Practice Questions

1. How does point-to-point differ from publish-subscribe?

Point-to-point delivers each message to exactly one consumer. Publish-subscribe delivers each message to all subscribed consumers. Point-to-point is for task distribution; pub-sub is for event broadcasting.

2. What happens if all consumers are busy?

Messages remain in the queue until a consumer becomes available. The queue depth increases. When a consumer finishes its current task, it picks up the next waiting message.

3. How do you ensure fair distribution across consumers?

Set prefetch_count=1 on the consumer channel. This tells the broker to send only one message at a time to each consumer, preventing one consumer from hoarding messages.

4. Can a point-to-point queue have multiple producers?

Yes. Multiple producers can send to the same queue. Each message is stored independently and delivered to exactly one consumer regardless of which producer sent it.

Challenge

Design a video transcoding system: videos are uploaded and queued. Ten worker instances each grab the next video. Transcoding takes 1-10 minutes per video. Implement prefetch, manual acknowledgments, and a dead-letter queue for videos that fail more than three times.

FAQ

Mini Project: Task Queue with Multiple Workers

import pika
import json
import threading
import time
import random

def start_worker(worker_id):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='work_tasks', durable=True)
    channel.basic_qos(prefetch_count=1)

    def callback(ch, method, properties, body):
        task = json.loads(body)
        print(f"[Worker {worker_id}] Processing task {task['id']}: {task['type']}")
        time.sleep(random.uniform(0.5, 2.0))
        print(f"[Worker {worker_id}] Completed task {task['id']}")
        ch.basic_ack(delivery_tag=method.delivery_tag)

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

for i in range(3):
    t = threading.Thread(target=start_worker, args=(i,), daemon=True)
    t.start()

time.sleep(1)

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

for i in range(10):
    task = {'id': i, 'type': random.choice(['encode', 'thumbnail', 'watermark'])}
    channel.basic_publish(
        exchange='',
        routing_key='work_tasks',
        body=json.dumps(task),
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"[Producer] Queued task {task['id']}: {task['type']}")

connection.close()
time.sleep(5)

Expected output:

[Producer] Queued task 0: encode
[Worker 0] Processing task 0: encode
[Producer] Queued task 1: thumbnail
[Worker 1] Processing task 1: thumbnail
[Producer] Queued task 2: watermark
[Worker 2] Processing task 2: watermark
...
[Worker 0] Completed task 0: encode
[Worker 0] Processing task 3: thumbnail

What's Next

Now that you understand point-to-point, explore the publish-subscribe pattern for broadcasting events to multiple consumers, or dive into message broker concepts for a deeper understanding of how brokers route messages.