Message Queue Implementation in Python: From Redis to Custom Sockets

Message queue implementation in Python starts with picking the right tool for your latency, durability, and scaling needs. I’ve run these in production for three years - here’s what actually works, what breaks, and when to roll your own.

Choosing the right message queue for your use case

What should I use if I need low latency and can tolerate brief downtime? Redis with its pub/sub or list-based patterns is often the sweet spot. It’s fast, simple, and integrates cleanly with FastAPI workers. But if you need guaranteed delivery or complex routing, RabbitMQ or Apache Kafka are better fits - though they add operational overhead. I’ve seen teams pick Kafka for event sourcing only to realize they needed a simple task queue; don’t over-engineer. For internal service-to-service comms where messages can be lost, Redis is fine. For financial transactions or user-facing workflows where loss means refunds or anger, go persistent.

Implementing a simple message queue with redis

How do I build a basic task queue using Redis in Python? Use Redis lists as a FIFO queue with BRPOPLPUSH for reliable processing. Here’s a minimal producer and consumer:

import redis
import time
import uuid

r = redis.Redis(host='localhost', port=6379, db=0)

def enqueue_task(task_data):
    task_id = str(uuid.uuid4())
    r.lpush('task_queue', f'{task_id}:{task_data}')
    return task_id

def process_tasks():
    while True:
        # BRPOPLPUSH moves item from source to dest and blocks until available
        item = r.brpoplpush('task_queue', 'processing_queue', timeout=0)
        if item:
            task_id, data = item.decode().split(':', 1)
            try:
                # Do work here
                result = process_data(data)
                r.lrem('processing_queue', 1, item)  # Remove from processing on success
                # Optionally store result
                r.set(f'result:{task_id}', result)
            except Exception as e:
                # On failure, push back to main queue for retry
                r.lpush('task_queue', item)
                r.lrem('processing_queue', 1, item)
                print(f"Task {task_id} failed: {e}")
        time.sleep(0.01)  # Prevent tight loop on error

def process_data(data):
    # Simulate work
    time.sleep(0.1)
    return f"processed_{data}"

This pattern gives you at-least-once delivery. If your worker crashes during processing, the item stays in processing_queue and can be recovered by a cleanup job. I’ve used this for image resizing pipelines - it’s survived network blips and worker restarts. But Redis isn’t durable by default; enable AOF or snapshots if you can’t lose tasks. Cost? A small Redis instance runs ~$5/month on most clouds. Worth it for most apps.

Building a message queue from scratch using sockets and threading

When would I ever need to roll my own message queue? Only if you’re learning, have extreme constraints, or need to embed queuing in a tiny device. For most backend work, don’t. But here’s how it works - so you understand what libraries abstract away.

import socket
import threading
import queue
import json

HOST = '127.0.0.1'
PORT = 65432
task_queue = queue.Queue()

def handle_client(conn, addr):
    with conn:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            try:
                task = json.loads(data.decode())
                task_queue.put(task)
                conn.sendall(b'ACK')
            except json.JSONDecodeError:
                conn.sendall(b'ERROR: Invalid JSON')

def worker():
    while True:
        task = task_queue.get()
        if task is None:  # Sentinel to shutdown
            break
        try:
            result = process_task(task)
            # In real system, send result via callback or DB
            print(f"Processed: {task} -> {result}")
        except Exception as e:
            print(f"Worker error: {e}")
        finally:
            task_queue.task_done()

def start_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((HOST, PORT))
        s.listen()
        print(f"Server listening on {HOST}:{PORT}")
        while True:
            conn, addr = s.accept()
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.start()

# Start workers
for _ in range(4):
    t = threading.Thread(target=worker)
    t.daemon = True
    t.start()

# Run server
try:
    start_server()
except KeyboardInterrupt:
    print("Shutting down...")
    task_queue.put(None)  # Signal workers to exit

This teaches you about backpressure, thread safety, and serialization - but it lacks persistence, clustering, and monitoring. I built a version like this for a sensor network where installing Redis wasn’t feasible. It worked for 50 devices. Scale past a hundred? Use a real broker. The failure mode here is silent: if the server dies, in-flight tasks are gone. No retries. No visibility. That’s why you don’t do this for anything critical.

Integrating celery with fastapi for async task processing

How do I offload long-running tasks in FastAPI without blocking requests? Celery is the battle-tested answer. It handles retries, scheduling, and worker management. Here’s how to wire it up:

from celery import Celery
from fastapi import FastAPI

celery_app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

@celery_app.task(bind=True, max_retries=3)
def process_heavy_task(self, data):
    try:
        # Simulate CPU or IO bound work
        result = heavy_computation(data)
        return result
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60, max_retries=3)

# main.py
from fastapi import FastAPI
from celery_worker import celery_app, process_heavy_task

app = FastAPI()

@app.post("/process/")
async def trigger_task(data: str):
    task = process_heavy_task.delay(data)
    return {"task_id": task.id, "status": "queued"}

@app.get("/result/{task_id}")
async def get_result(task_id: str):
    result = celery_app.AsyncResult(task_id)
    if result.ready():
        return {"task_id": task_id, "result": result.get()}
    return {"task_id": task_id, "status": "processing"}

I’ve used this for report generation and ML inference queues. The trade-off? Operational complexity. You now manage Redis, Celery workers, and monitoring. Workers can drift, queues can back up, and misconfigured timeouts cause zombie tasks. Monitor worker logs and queue lengths with Prometheus. If your tasks are under 200ms and infrequent, just use FastAPI’s background tasks. Celery shines when you need retries, prioritization, or scheduled jobs.

Handling message persistence and reliability in python message queues

What happens when my message queue loses data? With Redis, enable AOF with fsync everysec - it’s a good balance of speed and safety. For RabbitMQ, use durable queues and persistent messages (delivery_mode=2). In Celery, set task_acks_late=True and acks_on_failure_or_timeout=False so tasks aren’t removed until after they complete. I once lost a day’s worth of user uploads because a Redis restart flushed an unsaved RDB snapshot. Now I check: Is AOF on? Are consumers acknowledging? Do I have a dead-letter queue for failed messages? Test failure modes - kill a worker mid-task, unplug the network, simulate a broker restart. Your queue should survive.

Scaling message queues in distributed systems with python

How do I scale my Python message queue beyond a single node? Shard by task type or use consistent hashing for Redis clusters. For Celery, add more workers and monitor active and reserved counts via Flower. If using RabbitMQ, mirror queues across nodes or use quorum queues. I’ve scaled a Redis-backed queue to 50k msg/min by splitting queues per tenant and using Redis Cluster. Watch for hot keys - one tenant spamming tasks can starve others. Use rate limiting at the producer. For true scale, consider Kafka or Pulsar - they handle partitioning natively. But if you’re under 10k msg/min, a well-tuned Redis setup with multiple workers is simpler and cheaper.

FAQ

What is the simplest message queue implementation in Python for beginners? Start with Redis lists using LPUSH and BRPOP. It’s under 20 lines, requires only redis-py, and teaches core queuing concepts without broker complexity.

Can I use Python’s queue.Queue for inter-process communication? No. queue.Queue is thread-safe but not process-safe. For IPC, use multiprocessing.Queue or a broker like Redis - otherwise data won’t share across processes.

How do I prevent message loss in a Python message queue? Use persistent storage (AOF for Redis, durable queues for RabbitMQ), acknowledge messages after processing, and implement dead-letter queues for failed items.

When should I avoid using Celery with FastAPI? Avoid it for simple, fast background tasks (<100ms) where FastAPI’s BackgroundTasks suffices. Celery adds overhead that isn’t justified for trivial work.

Key Takeaways

• Match your queue choice to your durability and latency needs - don’t over-provision.
• Redis with BRPOPLPUSH gives you a reliable, simple queue for most Python apps.
• Custom socket-based queues are educational but lack production safety nets.
• Celery adds power for async tasks but introduces operational complexity.
• Test failure modes: assume your queue will lose connections, workers will crash, and brokers will restart.
• Monitor queue depth, worker latency, and retry rates - silent failures hurt more than loud ones.
• For under 10k msg/min, a single Redis instance with proper tuning often beats more complex solutions.