Exactly-once processing strategies in RabbitMQ - Mini Project: Build & Apply

Exactly-once Processing Strategies with RabbitMQ

📖 Scenario: You are building a simple message processing system using RabbitMQ. Your goal is to ensure that each message is processed exactly once, avoiding duplicates or message loss.

🎯 Goal: Learn how to set up a RabbitMQ queue with message acknowledgments and durable settings to achieve exactly-once processing behavior.

📋 What You'll Learn

Create a durable queue named task_queue

Set up a consumer that manually acknowledges messages

Simulate message processing with a delay

Print a confirmation after processing each message

💡 Why This Matters

🌍 Real World

Many applications need to process messages exactly once to avoid duplicate work or data corruption, such as order processing systems or financial transactions.

💼 Career

Understanding exactly-once processing with RabbitMQ is essential for DevOps engineers and backend developers working on reliable distributed systems.

1

Create a durable queue named task_queue

Write code to declare a durable queue called task_queue using RabbitMQ's channel method queue_declare with durable=True.

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# Declare a durable queue named 'task_queue'
# Your code here

Need a hint?

Use channel.queue_declare with queue='task_queue' and durable=True to create a queue that survives RabbitMQ restarts.

2

Set up a consumer with manual message acknowledgments

Add code to consume messages from task_queue with auto_ack=False to enable manual acknowledgments. Define a callback function named callback that takes ch, method, properties, and body as parameters.

RabbitMQ

import pika

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

# Define a callback function for consuming messages
# Your code here

Need a hint?

Define a function callback with the required parameters. Use channel.basic_consume with auto_ack=False to require manual acknowledgments.

3

Simulate message processing and send manual acknowledgment

Inside the callback function, add a 1-second delay to simulate processing using time.sleep(1). After processing, send a manual acknowledgment using ch.basic_ack(delivery_tag=method.delivery_tag). Import the time module at the top.

RabbitMQ

import pika
import time

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

def callback(ch, method, properties, body):
    print(f"Received {body.decode()}")
    # Simulate processing delay
    # Send manual acknowledgment
    # Your code here

channel.basic_consume(queue='task_queue', on_message_callback=callback, auto_ack=False)

Need a hint?

Use time.sleep(1) to simulate work. Call ch.basic_ack with the delivery tag to confirm message processing.

4

Start consuming and print confirmation after processing

Add code to start consuming messages with channel.start_consuming(). Also, inside the callback function, after acknowledgment, print "Done processing".

RabbitMQ

import pika
import time

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

def callback(ch, method, properties, body):
    print(f"Received {body.decode()}")
    time.sleep(1)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    # Print confirmation message
    # Your code here

channel.basic_consume(queue='task_queue', on_message_callback=callback, auto_ack=False)

# Start consuming messages
# Your code here

Need a hint?

Use print("Done processing") inside the callback after acknowledgment. Call channel.start_consuming() to begin message consumption.