Flow control mechanism in RabbitMQ - Mini Project: Build & Apply

Flow Control Mechanism with RabbitMQ

📖 Scenario: You are setting up a simple message queue system using RabbitMQ to handle tasks. To avoid overwhelming the consumer, you want to control the flow of messages so that the consumer processes only a limited number of messages at a time.

🎯 Goal: Build a Python script that connects to RabbitMQ, sets up a queue, configures flow control by limiting the number of unacknowledged messages the consumer can handle, and prints received messages.

📋 What You'll Learn

Create a connection to RabbitMQ server

Declare a queue named task_queue

Set the prefetch count to 1 to limit unacknowledged messages

Consume messages from the queue and print each message body

💡 Why This Matters

🌍 Real World

Flow control in message queues prevents consumers from being overwhelmed, ensuring stable and efficient processing of tasks.

💼 Career

Understanding flow control with RabbitMQ is essential for roles involving backend development, system integration, and DevOps to build reliable distributed systems.

1

Setup RabbitMQ connection and declare queue

Import pika and create a connection to RabbitMQ on localhost. Then create a channel and declare a queue named task_queue.

RabbitMQ

# Import pika and create connection
# Your code here

Need a hint?

Use pika.BlockingConnection with pika.ConnectionParameters('localhost') to connect. Use channel.queue_declare(queue='task_queue') to declare the queue.

2

Configure flow control with prefetch count

Set the channel's prefetch count to 1 using channel.basic_qos(prefetch_count=1) to limit the number of unacknowledged messages the consumer can receive.

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
# Set prefetch count to 1 for flow control
# Your code here

Need a hint?

Use channel.basic_qos(prefetch_count=1) to limit unacknowledged messages to one.

3

Define a callback function to process messages

Define a function named callback that takes ch, method, properties, and body as parameters. Inside the function, print the message body decoded as UTF-8, then acknowledge the message using ch.basic_ack(delivery_tag=method.delivery_tag).

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
channel.basic_qos(prefetch_count=1)

# Define callback function to process messages
# Your code here

Need a hint?

Define callback with the four parameters. Use print(f"Received {body.decode('utf-8')}") and acknowledge with ch.basic_ack(delivery_tag=method.delivery_tag).

4

Start consuming messages and print output

Use channel.basic_consume with the queue task_queue, the callback function, and auto_ack=False. Then start consuming with channel.start_consuming(). This will print received messages as they arrive.

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
channel.basic_qos(prefetch_count=1)

def callback(ch, method, properties, body):
    print(f"Received {body.decode('utf-8')}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Start consuming messages from 'task_queue'
# Your code here

Need a hint?

Use channel.basic_consume with queue='task_queue', on_message_callback=callback, and auto_ack=False. Then call channel.start_consuming().