Building a Resilient Task Queue System 📨 with RabbitMQ, Flask & Python 🐍

🐇 What is RabbitMQ?

In real world systems, services fail, crash, restart, or scale dynamically. You can't afford tight coupling between them.

RabbitMQ ensures no message gets lost, even if services aren't perfectly synced.

It’s a message broker a software that helps different parts of a system talk to each other using messages. It acts like a middleman which:

1. Receives tasks
2. Stores them safely in a queue
3. Delivers them to the right worker, even if the worker is offline at the time

🔌 Think of it like email for micro services.

📥 What is a Message Queue?

It gives you asynchronous communication and decoupling. That means:

A message queue is where tasks (messages) sit and wait to be picked up by a worker.

✅ Great for:

• Decoupling systems
• Load balancing
• Asynchronous processing

It’s foundational to scalable distributed systems.

It’s used in:

• 🛒 Order processing systems (e.g., Amazon, Shopify)
• 📧 Email/text notification pipelines
• 🧠 Microservices communication
• 🧾 Logging + audit trail systems

🐳 Running RabbitMQ with Docker

We used Docker to spin up RabbitMQ locally:

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

This exposes:

• 🔄 5672: App communication
• 📊 15672: Web dashboard — http://localhost:15672
• Login: guest / guest

🧰 What is Pika?

Pika is a Python client library for talking to RabbitMQ.

With Pika, you can:

• Connect to RabbitMQ
• Send (publish) messages to a queue
• Receive (consume) messages from a queue

We installed it using:

pip install pika

⚙️ Environment Setup

We used a Python virtual environment to manage dependencies:

python3 -m venv venv
source venv/bin/activate
pip install flask pika

💻 We Built Two Services

🧑‍🍳 Producer – Flask App

1. Accepts tasks via the /submit endpoint
2. Sends them to request_queue in RabbitMQ
3. Returns a correlation_id so clients can later check for results

📄 Code:

@app.route('/submit', methods=['POST'])

def submit_task():
  data = request.json
  correlation_id = str(uuid.uuid4())


  connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
  channel = connection.channel()
  channel.queue_declare(queue=REQUEST_QUEUE)
  channel.queue_declare(queue=REPLY_QUEUE)


  channel.basic_publish(
    exchange='',
    routing_key=REQUEST_QUEUE,
    body=json.dumps(data),
    properties=pika.BasicProperties(
      reply_to=REPLY_QUEUE,
      correlation_id=correlation_id
    )
  )
  connection.close()


return jsonify({"message": "Task submitted", "correlation_id": correlation_id}), 200

@app.route('/result/<correlation_id>', methods=['GET'])

def get_result(correlation_id):
 connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
 channel = connection.channel()
 channel.queue_declare(queue=REPLY_QUEUE)

 method_frame, properties, body = channel.basic_get(REPLY_QUEUE, auto_ack=False)

 while method_frame:
  if properties.correlation_id == correlation_id:
   channel.basic_ack(method_frame.delivery_tag)
   connection.close()
   return jsonify({"correlation_id": correlation_id, "result": json.loads(body)}), 200
  else:
   channel.basic_nack(method_frame.delivery_tag, requeue=True)
  method_frame, properties, body = channel.basic_get(REPLY_QUEUE, auto_ack=False)

 connection.close()
 return jsonify({"message": "Result not ready"}), 404

👷 Consumer – Worker Script

1. Listens to request_queue
2. Processes the task (e.g., a + b + c)
3. Sends result to reply_queue with the same correlation_id

📄 Code:

def callback(ch, method, properties, body):
 print("Received task:", body)
 data = json.loads(body)

 a = data.get('a', 0)
 b = data.get('b', 0)
 c = data.get('c', 0)
 result = a + b + c

 response = json.dumps({"result": result})

 ch.basic_publish(
  exchange='',
  routing_key=REPLY_QUEUE,
  body=response,
  properties=pika.BasicProperties(
   correlation_id=properties.correlation_id
  )
 )

 ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
channel = connection.channel()
channel.queue_declare(queue=REQUEST_QUEUE)
channel.queue_declare(queue=REPLY_QUEUE)
channel.basic_consume(queue=REQUEST_QUEUE, on_message_callback=callback)

print("Waiting for messages...")
channel.start_consuming()

🤕 What Happens if Services Crash?

❌ If the Producer Crashes:

1. No new tasks can be submitted
2. But existing messages in the queue are safe
3. ✅ When it's back online, task submission resumes

❌ If the Consumer Crashes:

1. Tasks sit in the queue untouched
2. When the consumer comes online, it processes all missed tasks

RabbitMQ makes the system resilient and fault-tolerant ✅

🧪 Example Usage

curl -X POST http://localhost:5050/submit \
-H "Content-Type: application/json" \
-d '{"a": 5, "b": 10, "c": 2}'

Response:

{ "correlation_id": "f40a1b9f-ef0f-4173-9444-f8c42f29b59c", 

"message": "Task submitted"}

Result:

curl http://localhost:5050/result/f40a1b9f-ef0f-4173-9444-f8c42f29b59c

{"correlation_id":"f40a1b9f-ef0f-4173-9444-f8c42f29b59c","result":{"result":17}}

Even when the consumer goes down you can still produce the task and queue it:

curl -X POST http://localhost:5050/submit -H "Content-Type: application/json" -d '{"a": 1, "b": 2, "c": 3}'

{"correlation_id":"3c5ac58e-5b8c-44ec-ad95-88a78f885215","message":"Task submitted"}

 curl -X POST http://localhost:5050/submit -H "Content-Type: application/json" -d '{"a": 10, "b": 5, "c": 6}'

{"correlation_id":"61fd1f8b-aead-4190-b837-a28fdd18efcb","message":"Task submitted"}