Today, i learnt about push vs pull architecture, the choice between push and pull architectures can significantly influence system performance, scalability, and user experience. Both approaches have their unique advantages and trade-offs. Understanding these architectures and their ideal use cases can help developers and architects make informed decisions.

What is Push Architecture?

Push architecture is a communication pattern where the server actively sends data to clients as soon as it becomes available. This approach eliminates the need for clients to repeatedly request updates.

How it Works

• The server maintains a connection with the client.
• When new data is available, the server “pushes” it to the connected clients.
• In a message queue context, producers send messages to a queue, and the queue actively delivers these messages to subscribed consumers without explicit requests.

Examples

• Notifications in Mobile Apps: Users receive instant updates, such as chat messages or alerts.
• Stock Price Updates: Financial platforms use push to provide real-time market data.
• Message Queues with Push Delivery: Systems like RabbitMQ or Kafka configured to push messages to consumers.
• Server-Sent Events (SSE) and WebSockets: These are common implementations of push.

Advantages

• Low Latency: Clients receive updates instantly, improving responsiveness.
• Reduced Redundancy: No need for clients to poll servers frequently, reducing bandwidth consumption.

Challenges

• Complexity: Maintaining open connections, especially for many clients, can be resource-intensive.
• Scalability: Requires robust infrastructure to handle large-scale deployments.

What is Pull Architecture?

Pull architecture involves clients actively requesting data from the server. This pattern is often used when real-time updates are not critical or predictable intervals suffice.

How it Works

• The client periodically sends requests to the server.
• The server responds with the requested data.
• In a message queue context, consumers actively poll the queue to retrieve messages when ready.

Examples

• Web Browsing: A browser sends HTTP requests to fetch pages and resources.
• API Data Fetching: Applications periodically query APIs to update information.
• Message Queues with Pull Delivery: Systems like SQS or Kafka where consumers poll for messages.
• Polling: Regularly checking a server or queue for updates.

Advantages

• Simpler Implementation: No need for persistent connections; standard HTTP requests or queue polling suffice.
• Server Load Control: The server can limit the frequency of client requests to manage resources better.

Challenges

• Latency: Updates are only received when the client requests them, which might lead to delays.
• Increased Bandwidth: Frequent polling can waste resources if no new data is available.

Yesterday, i came across a blog from inferable.ai https://www.inferable.ai/blog/posts/postgres-skip-locked, which walkthrough about using postgres as a queue. In this blog, i jot down notes on using postgres as a queue for future references.

PostgreSQL is a robust relational database that can be used for more than just storing structured data. With the SKIP LOCKED feature introduced in PostgreSQL 9.5, you can efficiently turn a PostgreSQL table into a job queue for distributed processing.

Why Use PostgreSQL as a Queue?

Using PostgreSQL as a queue can be advantageous because,

• Familiarity: If you’re already using PostgreSQL, there’s no need for an additional message broker.
• Durability: PostgreSQL ensures ACID compliance, offering reliability for your job processing.
• Simplicity: No need to manage another component like RabbitMQ or Kafka

Implementing a Queue with SKIP LOCKED

1. Create a Queue Table

To start, you need a table to store the jobs,

CREATE TABLE job_queue (
    id SERIAL PRIMARY KEY,
    job_data JSONB NOT NULL,
    status TEXT DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

This table has the following columns,

• id: A unique identifier for each job.
• job_data: The data or payload for the job.
• status: Tracks the job’s state (‘pending’, ‘in_progress’, or ‘completed’).
• created_at: Timestamp of job creation.

2. Insert Jobs into the Queue

Adding jobs is straightforward,

INSERT INTO job_queue (job_data)
VALUES ('{"task": "send_email", "email": "user@example.com"}');

3. Fetch Jobs for Processing with SKIP LOCKED

Workers will fetch jobs from the queue using SELECT ... FOR UPDATE SKIP LOCKED to avoid contention,

WITH next_job AS (
    SELECT id, job_data
    FROM job_queue
    WHERE status = 'pending'
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE job_queue
SET status = 'in_progress'
FROM next_job
WHERE job_queue.id = next_job.id
RETURNING job_queue.id, job_queue.job_data;

Key Points:

• FOR UPDATE locks the selected row to prevent other workers from picking it up.
• SKIP LOCKED ensures locked rows are skipped, enabling concurrent workers to operate without waiting.
• LIMIT 1 processes one job at a time per worker.

4. Mark Jobs as Completed

Once a worker finishes processing a job, it should update the job’s status,

UPDATE job_queue
SET status = 'completed'
WHERE id = $1; -- Replace $1 with the job ID

5. Delete Old or Processed Jobs

To keep the table clean, you can periodically remove completed jobs,

DELETE FROM job_queue
WHERE status = 'completed' AND created_at < NOW() - INTERVAL '30 days';

Example Worker Implementation

Here’s an example of a worker implemented in Python using psycopg2

import psycopg2
from psycopg2.extras import RealDictCursor

connection = psycopg2.connect("dbname=yourdb user=youruser")

while True:
    with connection.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(
            """
            WITH next_job AS (
                SELECT id, job_data
                FROM job_queue
                WHERE status = 'pending'
                FOR UPDATE SKIP LOCKED
                LIMIT 1
            )
            UPDATE job_queue
            SET status = 'in_progress'
            FROM next_job
            WHERE job_queue.id = next_job.id
            RETURNING job_queue.id, job_queue.job_data;
            """
        )

        job = cursor.fetchone()
        if job:
            print(f"Processing job {job['id']}: {job['job_data']}")

            # Simulate job processing
            cursor.execute("UPDATE job_queue SET status = 'completed' WHERE id = %s", (job['id'],))

        else:
            print("No jobs available. Sleeping...")
            time.sleep(5)

    connection.commit()

Considerations

1. Transaction Isolation: Use the REPEATABLE READ or SERIALIZABLE isolation level cautiously to avoid unnecessary locks.
2. Row Locking: SKIP LOCKED only skips rows locked by other transactions, not those locked within the same transaction.
3. Performance: Regularly archive or delete old jobs to prevent the table from growing indefinitely. Consider indexing the status column to improve query performance.
4. Fault Tolerance: Ensure that workers handle crashes or timeouts gracefully. Use a timeout mechanism to revert jobs stuck in the ‘in_progress’ state.
5. Scaling: Distribute workers across multiple nodes to handle a higher job throughput.
6. The SKIP LOCKED clause only applies to row-level locks – the required ROW SHARE table-level lock is still taken normally.
7. Using SKIP LOCKED provides an inconsistent view of the data by design. This is why it’s perfect for queue-like tables where we want to distribute work, but not suitable for general purpose work where consistency is required.

Today i learnt about Choreography pattern, where each and every service is communicating using a messaging queue. In this blog, i jot down notes on choreography pattern for my future self.

What is the Choreography Pattern?

In the Choreography Pattern, services communicate directly with each other via asynchronous events, without a central controller. Each service is responsible for a specific part of the workflow and responds to events produced by other services. This pattern allows for a more autonomous and loosely coupled system.

Key Features

• High scalability and independence of services.
• Decentralized control.
• Services respond to events they subscribe to.

When to Use the Choreography Pattern

• Event-Driven Systems: When workflows can be modeled as events triggering responses.
• High Scalability: When services need to operate independently and scale autonomously.
• Loose Coupling: When minimizing dependencies between services is critical.

Benefits of the Choreography Pattern

1. Decentralized Control: No single point of failure or bottleneck.
2. Increased Flexibility: Services can be added or modified without affecting others.
3. Better Scalability: Services operate independently and scale based on their workloads.
4. Resilience: The system can handle partial failures more gracefully, as services continue independently.

Example: E-Commerce Order Fulfillment

Problem

A fictional e-commerce platform needs to manage the following workflow:

1. Accepting an order.
2. Validating payment.
3. Reserving inventory.
4. Sending notifications to the customer.

Each step is handled by an independent service.

Solution

Using the Choreography Pattern, each service listens for specific events and publishes new events as needed. The workflow emerges naturally from the interaction of these services.

Implementation

Step 1: Define the Workflow as Events

• OrderPlaced: Triggered when a customer places an order.
• PaymentProcessed: Triggered after successful payment.
• InventoryReserved: Triggered after reserving inventory.
• NotificationSent: Triggered when the customer is notified.

Step 2: Implement Services

Each service subscribes to events and performs its task.

shared_utility.py

import pika
import json

def publish_event(exchange, event_type, data):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange, exchange_type='fanout')
    message = json.dumps({"event_type": event_type, "data": data})
    channel.basic_publish(exchange=exchange, routing_key='', body=message)
    connection.close()

def subscribe_to_event(exchange, callback):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange, exchange_type='fanout')
    queue = channel.queue_declare('', exclusive=True).method.queue
    channel.queue_bind(exchange=exchange, queue=queue)
    channel.basic_consume(queue=queue, on_message_callback=callback, auto_ack=True)
    print(f"Subscribed to events on exchange '{exchange}'")
    channel.start_consuming()

Order Service

from shared_utils import publish_event

def place_order(order_id, customer):
    print(f"Placing order {order_id} for {customer}")
    publish_event("order_exchange", "OrderPlaced", {"order_id": order_id, "customer": customer})

if __name__ == "__main__":
    # Simulate placing an order
    place_order(order_id=101, customer="John Doe")

Payment Service

from shared_utils import publish_event, subscribe_to_event
import time

def handle_order_placed(ch, method, properties, body):
    event = json.loads(body)
    if event["event_type"] == "OrderPlaced":
        order_id = event["data"]["order_id"]
        print(f"Processing payment for order {order_id}")
        time.sleep(1)  # Simulate payment processing
        publish_event("payment_exchange", "PaymentProcessed", {"order_id": order_id})

if __name__ == "__main__":
    subscribe_to_event("order_exchange", handle_order_placed)

Inventory Service

from shared_utils import publish_event, subscribe_to_event
import time

def handle_payment_processed(ch, method, properties, body):
    event = json.loads(body)
    if event["event_type"] == "PaymentProcessed":
        order_id = event["data"]["order_id"]
        print(f"Reserving inventory for order {order_id}")
        time.sleep(1)  # Simulate inventory reservation
        publish_event("inventory_exchange", "InventoryReserved", {"order_id": order_id})

if __name__ == "__main__":
    subscribe_to_event("payment_exchange", handle_payment_processed)

Notification Service

from shared_utils import subscribe_to_event
import time

def handle_inventory_reserved(ch, method, properties, body):
    event = json.loads(body)
    if event["event_type"] == "InventoryReserved":
        order_id = event["data"]["order_id"]
        print(f"Notifying customer for order {order_id}")
        time.sleep(1)  # Simulate notification
        print(f"Customer notified for order {order_id}")

if __name__ == "__main__":
    subscribe_to_event("inventory_exchange", handle_inventory_reserved)

Step 3: Run the Workflow

1. Start RabbitMQ using Docker as described above.
2. Run the services in the following order:
   • Notification Service: python notification_service.py
   • Inventory Service: python inventory_service.py
   • Payment Service: python payment_service.py
   • Order Service: python order_service.py
3. Place an order by running the Order Service. The workflow will propagate through the services as events are handled.

Key Considerations

1. Event Bus: Use an event broker like RabbitMQ, Kafka, or AWS SNS to manage communication between services.
2. Event Versioning: Include versioning to handle changes in event formats over time.
3. Idempotency: Ensure services handle repeated events gracefully to avoid duplication.
4. Monitoring and Tracing: Use tools like OpenTelemetry to trace and debug distributed workflows.
5. Error Handling:
   • Dead Letter Queues (DLQs) to capture failed events.
   • Retries with backoff for transient errors.

Advantages of the Choreography Pattern

1. Loose Coupling: Services interact via events without direct knowledge of each other.
2. Resilience: Failures in one service don’t block the entire workflow.
3. High Autonomy: Services operate independently and can be deployed or scaled separately.
4. Dynamic Workflows: Adding new services to the workflow requires subscribing them to relevant events.

Challenges of the Choreography Pattern

1. Complex Debugging: Tracing errors across distributed services can be difficult.
2. Event Storms: Poorly designed workflows may generate excessive events, overwhelming the system.
3. Coordination Overhead: Decentralized logic can lead to inconsistent behavior if not carefully managed.

Orchestrator vs. Choreography: When to Choose?

• Use Orchestrator Pattern when workflows are complex, require central control, or involve many dependencies.
• Use Choreography Pattern when you need high scalability, loose coupling, or event-driven workflows.

Today, i learnt about Queue Based Loading pattern, which helps to manage intermittent peak load to a service via queues. Basically decoupling Tasks from Services. In this blog i jot down notes on this pattern for my future self.

In today’s digital landscape, applications are expected to handle large-scale operations efficiently. Whether it’s processing massive data streams, ensuring real-time responsiveness, or integrating with multiple third-party services, scalability and reliability are paramount. One pattern that elegantly addresses these challenges is the Queue-Based Loading Pattern.

What Is the Queue-Based Loading Pattern?

The Queue-Based Loading Pattern leverages message queues to decouple and coordinate tasks between producers (such as applications or services generating data) and consumers (services or workers processing that data). By using queues as intermediaries, this pattern allows systems to manage workloads efficiently, ensuring seamless and scalable operation.

Key Components of the Pattern

1. Producers: Producers are responsible for generating tasks or data. They send these tasks to a message queue instead of directly interacting with consumers. Examples include:
   • Web applications logging user activity.
   • IoT devices sending sensor data.
2. Message Queue: The queue acts as a buffer, storing tasks until consumers are ready to process them. Popular tools for implementing queues include RabbitMQ, Apache Kafka, AWS SQS, and Redis.
3. Consumers: Consumers retrieve messages from the queue and process them asynchronously. They are typically designed to handle tasks independently and at their own pace.
4. Processing Logic: This is the core functionality that processes the tasks retrieved by consumers. For example, resizing images, sending notifications, or updating a database.

How It Works

1. Task Generation: Producers push tasks to the queue as they are generated.
2. Message Storage: The queue stores tasks in a structured manner (FIFO, priority-based, etc.) and ensures reliable delivery.
3. Task Consumption: Consumers pull tasks from the queue, process them, and optionally acknowledge completion.
4. Scalability: New consumers can be added dynamically to handle increased workloads, ensuring the system remains responsive.

Benefits of the Queue-Based Loading Pattern

1. Decoupling: Producers and consumers operate independently, reducing tight coupling and improving system maintainability.
2. Scalability: By adding more consumers, systems can easily scale to handle higher workloads.
3. Fault Tolerance: If a consumer fails, messages remain in the queue, ensuring no data is lost.
4. Load Balancing: Tasks are distributed evenly among consumers, preventing any single consumer from becoming a bottleneck.
5. Asynchronous Processing: Consumers can process tasks in the background, freeing producers to continue generating data without delay.

Issues and Considerations

1. Rate Limiting: Implement logic to control the rate at which services handle messages to prevent overwhelming the target resource. Test the system under load and adjust the number of queues or service instances to manage demand effectively.
2. One-Way Communication: Message queues are inherently one-way. If tasks require responses, you may need to implement a separate mechanism for replies.
3. Autoscaling Challenges: Be cautious when autoscaling consumers, as it can lead to increased contention for shared resources, potentially reducing the effectiveness of load leveling.
4. Traffic Variability: Consider the variability of incoming traffic to avoid situations where tasks pile up faster than they are processed, creating a perpetual backlog.
5. Queue Persistence: Ensure your queue is durable and capable of persisting messages. Crashes or system limits could lead to dropped messages, risking data loss.

Use Cases

1. Email and Notification Systems: Sending bulk emails or push notifications without overloading the main application.
2. Data Pipelines: Ingesting, transforming, and analyzing large datasets in real-time or batch processing.
3. Video Processing: Queues facilitate tasks like video encoding and thumbnail generation.
4. Microservices Communication: Ensures reliable and scalable communication between microservices.

Best Practices

1. Message Durability: Configure your queue to persist messages to disk, ensuring they are not lost during system failures.
2. Monitoring and Metrics: Use monitoring tools to track queue lengths, processing rates, and consumer health.
3. Idempotency: Design consumers to handle duplicate messages gracefully.
4. Error Handling and Dead Letter Queues (DLQs): Route failed messages to DLQs for later analysis and reprocessing.

Today, i learnt about competing consumer, its a simple concept of consuming messages with many consumers. In this blog, i jot down notes on competing consumer for better understanding.

The competing consumer pattern is a commonly used design paradigm in distributed systems for handling workloads efficiently. It addresses the challenge of distributing tasks among multiple consumers to ensure scalability, reliability, and better resource utilization. In this blog, we’ll delve into the details of this pattern, its implementation, and its benefits.

What is the Competing Consumer Pattern?

The competing consumer pattern involves multiple consumers that independently compete to process messages or tasks from a shared queue. This pattern is particularly effective in scenarios where the rate of incoming tasks is variable or high, as it allows multiple consumers to process tasks concurrently.

Key Components

1. Producer: The component that generates tasks or messages.
2. Queue: A shared storage medium (often a message broker) that holds tasks until a consumer is ready to process them.
3. Consumer: The component that processes tasks. Multiple consumers operate concurrently and compete for tasks in the queue.
4. Message Broker: Middleware (e.g., RabbitMQ, Kafka) that manages the queue and facilitates communication between producers and consumers.

How It Works (Message as Tasks)

1. Task Generation
   • Producers create tasks and push them into the queue.
   • Tasks can represent anything, such as processing an image, sending an email, or handling a database operation.
2. Task Storage
   • The queue temporarily stores tasks until they are picked up by consumers.
   • Queues often support features like message persistence and delivery guarantees to enhance reliability.
3. Task Processing
   • Consumers pull tasks from the queue and process them independently.
   • Each consumer works on one task at a time, and no two consumers process the same task simultaneously.
4. Task Completion
   • Upon successful processing, the consumer acknowledges the task’s completion to the message broker.
   • The message broker then removes the task from the queue.

Handling Poison Messages

A poison message is a task or message that a consumer repeatedly fails to process. Poison messages can cause delays, block the queue, or crash consumers if not handled appropriately.

Strategies for Handling Poison Messages

1. Retry Mechanism
   • Allow a fixed number of retries for a task before marking it as failed.
   • Use exponential backoff to reduce the load on the system during retries.
2. Dead Letter Queue (DLQ)
   • Configure a Dead Letter Queue to store messages that cannot be processed after a predefined number of attempts.
   • Poison messages in the DLQ can be analyzed or reprocessed manually.
3. Logging and Alerting
   • Log details about the poison message for further debugging.
   • Set up alerts to notify administrators when a poison message is encountered.
4. Idempotent Consumers
   • Design consumers to handle duplicate processing gracefully. This prevents issues if a message is retried multiple times.

RabbitMQ Example

Producer

import pika

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

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

messages = ["Task 1", "Task 2", "Task 3"]

for message in messages:
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2,  # Makes the message persistent
        )
    )
    print(f"[x] Sent {message}")

connection.close()

Dead Letter Exchange

channel.queue_declare(queue='task_queue', durable=True, arguments={
    'x-dead-letter-exchange': 'dlx_exchange'
})
channel.exchange_declare(exchange='dlx_exchange', exchange_type='fanout')
channel.queue_declare(queue='dlq', durable=True)
channel.queue_bind(exchange='dlx_exchange', queue='dlq')

Consumer Code

import pika
import time

def callback(ch, method, properties, body):
    try:
        print(f"[x] Received {body}")
        # Simulate task processing
        if body == b"Task 2":
            raise ValueError("Cannot process this message")
        time.sleep(1)
        print(f"[x] Processed {body}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"[!] Failed to process message: {body}, error: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

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

channel.queue_declare(queue='task_queue', durable=True)
print('[*] Waiting for messages. To exit press CTRL+C')

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

channel.start_consuming()

Benefits of the Competing Consumer Pattern

1. Scalability Adding more consumers allows the system to handle higher workloads.
2. Fault Tolerance If a consumer fails, other consumers can continue processing tasks.
3. Resource Optimization Consumers can be distributed across multiple machines to balance the load.
4. Asynchronous Processing Decouples task generation from task processing, enabling asynchronous workflows.

Challenges and Considerations

1. Message Duplication – In some systems, messages may be delivered more than once. Implement idempotent processing to handle duplicates.
2. Load Balancing – Ensure tasks are evenly distributed among consumers to avoid bottlenecks.
3. Queue Overload – High task rates may lead to queue overflow. Use rate limiting or scale your infrastructure to prevent this.
4. Monitoring and Metrics – Implement monitoring to track queue sizes, processing rates, and consumer health.
5. Poison Messages – Implement a robust strategy for handling poison messages, such as using a DLQ or retry mechanism.

References

1. https://www.enterpriseintegrationpatterns.com/patterns/messaging/CompetingConsumers.html
2. https://dev.to/willvelida/the-competing-consumers-pattern-4h5n
3. https://medium.com/event-driven-utopia/competing-consumers-pattern-explained-b338d54eff2b

What Are Lazy Queues?

• Lazy Queues are designed to store messages primarily on disk rather than in memory.
• They are optimized for use cases involving large message backlogs where minimizing memory usage is critical.

Key Characteristics

1. Disk-Based Storage – Messages are stored on disk immediately upon arrival, rather than being held in memory.
2. Low Memory Usage – Only minimal metadata for messages is kept in memory.
3. Scalability – Can handle millions of messages without consuming significant memory.
4. Message Retrieval – Retrieving messages is slower because messages are fetched from disk.
5. Durability – Messages persist on disk, reducing the risk of data loss during RabbitMQ restarts.

Trade-offs

• Latency: Fetching messages from disk is slower than retrieving them from memory.
• Throughput: Not suitable for high-throughput, low-latency applications.

Choose Lazy Queues if

• You need to handle very large backlogs of messages.
• Memory is a constraint in your system.Latency and throughput are less critical.

Implementation

Pre-requisites

1. Install and run RabbitMQ on your local machine.

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0-management

2. Install the pika library

pip install pika

Producer (producer.py)

This script sends a persistent message to a Lazy Queue.

import pika

# RabbitMQ connection parameters for localhost
connection_params = pika.ConnectionParameters(host="localhost")

# Connect to RabbitMQ
connection = pika.BlockingConnection(connection_params)
channel = connection.channel()

# Custom Exchange and Routing Key
exchange_name = "custom_exchange"
routing_key = "custom_routing_key"
queue_name = "lazy_queue_example"

# Declare the custom exchange
channel.exchange_declare(
    exchange=exchange_name,
    exchange_type="direct",  # Direct exchange routes messages based on the routing key
    durable=True
)

# Declare a Lazy Queue
channel.queue_declare(
    queue=queue_name,
    durable=True,
    arguments={"x-queue-mode": "lazy"}  # Configure the queue as lazy
)

# Bind the queue to the custom exchange with the routing key
channel.queue_bind(
    exchange=exchange_name,
    queue=queue_name,
    routing_key=routing_key
)

# Publish a message
message = "Hello from the Producer via Custom Exchange!"
channel.basic_publish(
    exchange=exchange_name,
    routing_key=routing_key,
    body=message,
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent message
)

print(f"Message sent to Lazy Queue via Exchange: {message}")

# Close the connection
connection.close()

Consumer (consumer.py)

import pika

# RabbitMQ connection parameters for localhost
connection_params = pika.ConnectionParameters(host="localhost")

# Connect to RabbitMQ
connection = pika.BlockingConnection(connection_params)
channel = connection.channel()

# Custom Exchange and Routing Key
exchange_name = "custom_exchange"
routing_key = "custom_routing_key"
queue_name = "lazy_queue_example"

# Declare the custom exchange
channel.exchange_declare(
    exchange=exchange_name,
    exchange_type="direct",  # Direct exchange routes messages based on the routing key
    durable=True
)

# Declare the Lazy Queue
channel.queue_declare(
    queue=queue_name,
    durable=True,
    arguments={"x-queue-mode": "lazy"}  # Configure the queue as lazy
)

# Bind the queue to the custom exchange with the routing key
channel.queue_bind(
    exchange=exchange_name,
    queue=queue_name,
    routing_key=routing_key
)

# Callback function to process messages
def callback(ch, method, properties, body):
    print(f"Received message: {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)  # Acknowledge the message

# Start consuming messages
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=False)

print("Waiting for messages. To exit, press CTRL+C")
try:
    channel.start_consuming()
except KeyboardInterrupt:
    print("Stopped consuming.")

# Close the connection
connection.close()

Explanation

1. Producer
   • Defines a custom exchange (custom_exchange) of type direct.
   • Declares a Lazy Queue (lazy_queue_example).
   • Binds the queue to the exchange using a routing key (custom_routing_key).
   • Publishes a persistent message via the custom exchange and routing key.
2. Consumer
   • Declares the same exchange and Lazy Queue to ensure they exist.
   • Consumes messages routed to the queue through the custom exchange and routing key.
3. Custom Exchange and Binding
   • The direct exchange type routes messages based on an exact match of the routing key.
   • Binding ensures the queue receives messages published to the exchange with the specified key.
4. Lazy Queue Behavior
   • Messages are stored directly on disk to minimize memory usage.

What Are Quorum Queues?

• Quorum Queues are distributed queues built on the Raft consensus algorithm.
• They are designed for high availability, durability, and data safety by replicating messages across multiple nodes in a RabbitMQ cluster.
• Its a replacement of Mirrored Queues.

Key Characteristics

1. Replication:
   • Messages are replicated across a quorum (a majority of nodes).
   • A quorum consists of an odd number of replicas (e.g., 3, 5, 7) to ensure a majority can elect a leader during failovers.
2. Leader-Follower Architecture:
   • Each Quorum Queue has one leader and multiple followers.
   • The leader handles all write and read operations, while followers replicate messages and provide redundancy.
3. Durability:
   • Messages are written to disk on all quorum nodes, ensuring persistence even if nodes fail.
4. High Availability:
   • If the leader node fails, RabbitMQ elects a new leader from the remaining quorum, ensuring continued operation.
5. Consistency:
   • Quorum Queues prioritize consistency over availability.
   • Messages are acknowledged only after replication is successful on a majority of nodes.
6. Message Ordering:
   • Message ordering is preserved during normal operations but may be disrupted during leader failovers.

Use Cases

• Mission-Critical Applications – Systems where message loss is unacceptable (e.g., financial transactions, order processing).
• Distributed Systems – Environments requiring high availability and fault tolerance.
• Data Safety – Applications prioritizing consistency over throughput (e.g., event logs, audit trails).

Setups

Using rabbitmqctl

rabbitmqctl add_queue quorum_queue --type quorum

Using python

channel.queue_declare(queue="quorum_queue", arguments={"x-queue-type": "quorum"})

References:

1. https://www.rabbitmq.com/docs/quorum-queues

Github: https://github.com/syedjaferk/rabbitmq_message_queues

Imagine you own a busy online store. Customers place orders, payments are processed, inventory is updated, and confirmation emails are sent.

If these steps happen one after another in real-time (synchronous), your website could slow down or even crash under high demand. This is where message queues come in to picture. They help different parts of your system communicate smoothly and handle tasks efficiently, even during a rush. Its one of the solution for asynchronous communication.

What is a Message Queue?

A message queue is a software system that enables different parts of an application to send and receive messages asynchronously. Messages are temporarily stored in a queue until the recipient is ready to process them.

For example, think of it as a waiting line at a busy coffee shop. Each order (or message) waits in line until it’s picked up and handled by a coffee maker (or worker). The beauty of a message queue is that the coffee shop (producer) can keep taking orders without waiting for the coffee maker (consumer) to finish the current one.

Here’s how it works:

• The producer sends messages to the queue.
• The queue stores the messages.
• The consumer picks up messages one by one to process them.

RabbitMQ is one kind of tool which helps in enabling async communication.

Key Components of RabbitMQ (a Popular Message Queue System)

1. Producer: The sender of messages. For example, your website sending an order to the queue.
2. Queue: The holding area for messages, like a to-do list. Each order waits here until processed.
3. Consumer: The worker that processes messages. For example, the service that charges a credit card.
4. Exchange: Think of this as a traffic controller. It decides which queue gets each message based on rules you set.
5. Message: The data being sent, such as order details (customer name, items, total price).
6. Acknowledgements (ACKs): A signal from the consumer to RabbitMQ saying, “Message processed successfully!”.

How a Message Queue Solves Real Problems

Scenario: Imagine your online store uses a message queue during a holiday rush.

1. Placing Orders
   • Customers place orders on your website (producer).
   • Orders are sent to the RabbitMQ queue.
2. Processing Payments
   • The payment service (consumer) picks up orders from the queue, one by one, to charge credit cards.
3. Sending Emails
   • Once payment is successful, another consumer sends confirmation emails.
4. Updating Inventory
   • A third consumer updates the inventory system.

Without a queue: All these tasks would happen one after the other, causing delays and potential failures.

With a queue: Each task works independently and efficiently, ensuring smooth operations.

Simple RabbitMQ Example

Step1: I am spinning up a RabbitMQ from a Docker

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0-management

Step 2: Producer Code (Sending Messages)

import pika

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

channel.queue_declare(queue='order_queue')

channel.basic_publish(exchange='',
                      routing_key='order_queue',
                      body='Order #12345')
print("[x] Sent 'Order #12345'")
connection.close()

Explanation

1. pika.ConnectionParameters('localhost'): Connects to RabbitMQ running locally.
2. channel.queue_declare(queue='order_queue'): Ensures the queue exists. If it doesn’t, RabbitMQ will create it.
3. channel.basic_publish(...): Publishes a message (in this case, “Order #12345”) to the specified queue.
4. connection.close(): Cleans up and closes the connection.

Step 3: Consumer Code (Processing Message)

import pika

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

channel.queue_declare(queue='order_queue')

def callback(ch, method, properties, body):
    print(f"[x] Processed {body}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='order_queue', on_message_callback=callback)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

Explanation

1. channel.queue_declare(queue='order_queue'): Ensures the consumer is listening to the correct queue.
2. callback: A function that processes each message. Here, it prints the message content and acknowledges it.
3. channel.basic_consume(...): Binds the callback function to the queue, so the consumer processes messages as they arrive.
4. channel.start_consuming(): Starts the consumer, waiting for messages indefinitely.

Best Practices (Not Tried – Just Got it from Course page.)

1. Keep Messages Small: Only send necessary data to avoid delays.
2. Use Dead Letter Queues: Handle failed messages separately to keep the main queue clear.
3. Monitor Performance: Watch queue sizes and processing times to prevent backlogs.
4. Scale Consumers: Add more workers during busy times to process messages faster.
5. Secure Your System: Use encryption and authentication to protect sensitive data.