How the Producer-Consumer Pattern Is Implemented in Distributed Systems Using Message Queues

The producer-consumer pattern in distributed systems relies on a message queue to decouple senders from receivers through either point-to-point delivery (exactly one consumer) or publish-subscribe broadcasting (multiple consumers).

According to the CyC2018/CS-Notes repository, message queues serve as the backbone of asynchronous communication in distributed architectures, enabling horizontal scaling and fault tolerance while abstracting the complexity of direct service-to-service calls. This implementation leverages two fundamental models that define how messages flow between producers and consumers.

Core Messaging Models

The repository outlines two primary architectures that determine message delivery semantics and consumer behavior.

Point-to-Point (点对点)

In the point-to-point model, a producer pushes a message into a queue where exactly one consumer retrieves and processes it. The queue guarantees that each message is consumed once, making this ideal for asynchronous task processing such as sending verification emails after user registration. As detailed in notes/消息队列.md, this model ensures that work is distributed evenly across consumer instances without duplication.

Publish-Subscribe (发布/订阅)

The publish-subscribe model enables a producer to publish messages to a topic or channel, allowing multiple consumers to subscribe and each receive an independent copy of the message. This pattern supports event broadcasting, cache invalidation, and log collection scenarios where multiple services must react to the same event simultaneously.

Architectural Implementation Steps

According to notes/消息队列.md and notes/分布式.md, implementing this pattern involves six distinct phases that ensure reliable message flow from producer to consumer.

1. Message Creation and Serialization

The producer formats a payload (JSON, protobuf, or Avro) and invokes the queue's produce API (e.g., publish, send, or push). The message typically includes metadata such as event type, timestamp, and unique identifiers to support deduplication downstream.

2. Broker Persistence

The message broker stores the message reliably using disk write-ahead logs or replication across nodes. This persistence prevents data loss during producer or broker failures before the consumer acknowledges receipt.

3. Delivery Semantics Configuration

Depending on the broker configuration, the system can guarantee at-least-once, at-most-once, or exactly-once delivery. Most distributed systems opt for at-least-once delivery combined with idempotent consumers, as exactly-once semantics incur higher latency and complexity.

4. Message Consumption and Acknowledgment

Consumers either poll the broker or receive push notifications. After processing, the consumer must explicitly acknowledge successful handling via an ack() call. If processing fails, the consumer issues a nack() (negative acknowledgment) to trigger requeueing or dead-letter routing.

5. Horizontal Scaling

Multiple producer instances write concurrently to the same topic or queue, while consumer groups enable parallel processing. In point-to-point systems, competing consumers read from the same queue; in pub-sub systems, consumer groups allow each instance to process a subset of partitions (as implemented in Kafka).

6. Reliability Enhancements

Idempotent handling prevents duplicate effects during retries by checking deduplication tables or unique message IDs before processing. The repository emphasizes that consumers should implement 幂等性 (idempotency) checks to handle the "本地消息表 + 消息队列" (local message table + message queue) pattern described in notes/分布式.md, ensuring eventual consistency across distributed transactions.

Practical Code Examples

These language-agnostic snippets illustrate the core patterns found in notes/消息队列.md and notes/Redis.md. Replace placeholder methods with your specific broker API (Kafka, RabbitMQ, or Redis).

Producer Implementation

import json
import time

# 1️⃣ Create message payload with unique ID for deduplication

payload = {
    "event": "user_registered",
    "event_id": "evt_123456789",  # Unique ID for idempotency checks

    "user_id": 12345,
    "timestamp": int(time.time())
}

# 2️⃣ Serialize payload

msg = json.dumps(payload)

# 3️⃣ Publish to topic or queue

# Kafka: producer.send('user-events', value=msg)

# RabbitMQ: channel.basic_publish(exchange='events', routing_key='user.registered', body=msg)

queue_client.publish(topic="user-events", message=msg)

Consumer Implementation with Idempotency

def handle_message(raw_msg):
    data = json.loads(raw_msg)
    
    # Idempotent processing: check deduplication store first

    if is_already_processed(data["event_id"]):
        return  # Skip duplicate

    
    # Business logic execution

    send_welcome_email(data["user_id"])
    
    # Mark as processed in deduplication table

    mark_as_processed(data["event_id"])

# Main consumption loop

for raw_msg in queue_client.consume(topic="user-events"):
    try:
        handle_message(raw_msg)
        # 2️⃣ Acknowledge successful processing

        queue_client.ack(raw_msg)
    except Exception:
        # 3️⃣ Requeue or send to dead-letter queue on failure

        queue_client.nack(raw_msg, requeue=True)

Point-to-Point Example Using Redis Lists

As noted in notes/Redis.md, Redis Lists can function as simple message queues using atomic push/pop operations, though production systems typically prefer dedicated brokers for persistence guarantees.

# Producer: Push task to left of list

redis-cli LPUSH user_task_queue "$(cat task.json)"

# Consumer: Atomic pop from right with polling loop

while true; do
    task=$(redis-cli RPOP user_task_queue)
    [ -z "$task" ] && sleep 1 && continue
    # Process $task idempotently...

    echo "Processing: $task"
done

Summary

• Message queues decouple producers from consumers, enabling asynchronous processing and fault tolerance in distributed systems.
• Point-to-point guarantees exactly-one consumption per message, while publish-subscribe broadcasts to multiple consumers.
• Implementation requires serialization, broker persistence, configurable delivery semantics, explicit acknowledgment, and horizontal scaling strategies.
• Reliability depends on idempotent consumer logic and deduplication mechanisms rather than exactly-once delivery guarantees.
• Source files notes/消息队列.md, notes/Redis.md, and notes/分布式.md provide the architectural foundation for these patterns in the CyC2018/CS-Notes repository.

Frequently Asked Questions

What is the difference between message queues and the observer pattern?

The observer pattern described in notes/设计模式 - 观察者.md operates within a single process for in-memory event notification, while message queues provide inter-process communication across distributed nodes with persistence guarantees. Message queues extend the observer concept to distributed systems by adding durability, delivery acknowledgments, and horizontal scaling capabilities that in-memory observers cannot provide.

Why do most systems use at-least-once delivery instead of exactly-once?

Exactly-once delivery requires complex distributed transaction coordination between producers, brokers, and consumers, significantly impacting throughput and latency. At-least-once delivery combined with idempotent consumers offers better performance while maintaining data consistency, as consumers simply check deduplication tables before processing duplicate messages.

How does Redis implement message queues compared to Kafka or RabbitMQ?

According to notes/Redis.md, Redis uses list commands (LPUSH/RPOP) or Pub/Sub commands for lightweight messaging, but lacks built-in persistence acknowledgments and consumer group management. Kafka and RabbitMQ provide durable retention, automatic rebalancing of consumer groups, and explicit acknowledgment protocols that Redis cannot guarantee, making dedicated brokers preferable for production distributed systems.

What prevents consumers from processing the same message twice?

Idempotent processing logic implemented at the consumer level prevents duplicate effects. Consumers store processed message IDs in a deduplication table (often using the same database transaction as the business logic) and skip processing if the ID already exists. This pattern, referenced in notes/消息队列.md under 接收端的可靠性, ensures exactly-once processing semantics even when the underlying queue provides only at-least-once delivery.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}