Python Message Queue Between Processes: IPC Options for FastAPI Workers

I use multiprocessing.Queue as a python message queue between processes every day in production FastAPI apps. It’s not flashy but it works when you need to move data between worker pools and async request handlers without adding external dependencies.

How do I implement inter-process communication with multiprocessing.Queue?

You create a Queue instance in your parent process, pass it to child processes via fork or spawn, then use .put() and .get() to exchange data. Here’s a minimal FastAPI background worker pattern:

from multiprocessing import Process, Queue
import time
from fastapi import FastAPI

def worker(q: Queue):
    while True:
        item = q.get()
        if item is None:  # Sentinel to stop
            break
        print(f"Processing {item}")
        time.sleep(0.1)  # Simulate work

app = FastAPI()
q = Queue()
p = Process(target=worker, args=(q,))
p.start()

@app.post("/task")
async def add_task(data: dict):
    q.put(data)
    return {"status": "queued"}

@app.on_event("shutdown")
def shutdown_event():
    q.put(None)
    p.join()

This keeps things simple: no Redis, no RabbitMQ, just built-in Python. I’ve used this pattern for CPU-bound tasks like image resizing or model inference where the GIL blocks asyncio.

What’s the difference between multiprocessing.Queue, pipes, and shared memory?

multiprocessing.Queue is thread- and process-safe, handles serialization automatically, and scales to multiple producers/consumers. Pipes are faster for two-endpoint communication but lack built-in buffering and can deadlock if not managed carefully. Shared memory (via Array or Value) is fastest for raw bytes but requires you to handle synchronization yourself with Locks or Semaphores.

In practice, I reach for Queue first. It’s the safest default. I only drop to pipes when I’m doing low-latency, two-way signaling between exactly two processes - like a control plane and a data plane. Shared memory? Only when I’m moving large numpy arrays and profiling shows serialization is the bottleneck. Even then, I wrap it in a Queue-like interface to avoid footguns.

How do I handle serialization of complex objects in process queues?

Pickle is the default serializer for multiprocessing.Queue. It works for most Python objects but fails with lambdas, local functions, or objects with non-picklable attributes (like open file handles or database connections). I’ve been bitten by this when trying to pass SQLAlchemy sessions across processes - don’t do that.

Instead, pass primitive types or simple dataclasses. If you need to move complex data, serialize it yourself first:

import pickle
from multiprocessing import Queue

def safe_put(q: Queue, obj):
    q.put(pickle.dumps(obj))

def safe_get(q: Queue):
    return pickle.loads(q.get())

This adds overhead but gives you control. For FastAPI workers doing ML inference, I often pass model input as numpy arrays (which pickle handles fine) and keep the model itself loaded in each worker process to avoid repeated serialization.

How do I avoid deadlocks and race conditions in multiprocessing queues?

Deadlocks happen when you forget to consume items or use Queue.join() incorrectly. I once had a worker that crashed silently, leaving the queue full - then the producer blocked forever on .put(). The fix? Always use timeouts and monitor queue size.

try:
    q.put(item, timeout=1.0)
except Full:
    logger.warning("Queue full, dropping task")
    # or retry, or alert

For race conditions: Queue’s internal locks make .put() and .get() safe, but if you’re doing check-then-act (like if not q.empty(): q.get()), you’re already broken. Never rely on .empty() or .qsize() for logic - those are approximate in multiprocessing contexts due to process scheduling.

I’ve seen teams try to build priority queues on top of Queue using these methods. It never ends well. Use a proper priority queue implementation or switch to Redis if ordering matters.

What are the performance benchmarks of different IPC mechanisms in Python?

In my local tests (Linux, Python 3.11, 8-core CPU):

multiprocessing.Queue: ~50k msg/s for small dicts
Pipe (duplex): ~200k msg/s
Shared memory (Array): ~500k msg/s
Redis (local): ~100k msg/s (network overhead)

But raw speed isn’t the whole story. Queue adds ~10µs latency per message vs ~2µs for pipes. For FastAPI background workers handling 1k req/s, that difference is negligible. What matters more is reliability and ease of debugging. I’ve seen shared memory implementations corrupt data when a worker segfaults mid-write. Queue just drops the message - annoying but recoverable.

How do I integrate message queues with FastAPI background workers?

I combine multiprocessing.Queue with FastAPI’s lifespan events and dependency injection. Here’s a cleaner version using a manager class:

from contextlib import asynccontextmanager
from multiprocessing import Process, Queue
from fastapi import FastAPI

class TaskQueue:
    def __init__(self):
        self.q = Queue()
        self.process = None

    def start(self):
        self.process = Process(target=self._worker, args=(self.q,))
        self.process.start()

    def _worker(self, q):
        while True:
            task = q.get()
            if task is None:
                break
            self._process_task(task)

    def _process_task(self, task):
        # Your actual work here
        pass

    def stop(self):
        if self.process:
            self.q.put(None)
            self.process.join()

@asynccontextmanager
async def lifespan(app: FastAPI):
    task_queue = TaskQueue()
    task_queue.start()
    yield {"task_queue": task_queue}
    task_queue.stop()

app = FastAPI(lifespan=lifespan)

@app.post("/task")
async def add_task(data: dict, request: Request):
    request.state.task_queue.q.put(data)
    return {"status": "queued"}

This keeps the queue lifecycle tied to the app. I’ve used this in RAG pipelines where the worker does embedding generation and the API just enqueues text chunks. For more on choosing between built-in queues and external brokers, see my post on choosing a python message queue library for production.

When NOT to use multiprocessing.Queue

Don’t use it if:

You need persistence across reboots (use Redis or RabbitMQ)
You’re crossing machine boundaries (use a network queue)
You require strict ordering or transactions (Queue gives none of these)
Your workers are mostly I/O-bound (stick with asyncio and background tasks)

I’ve seen teams try to scale this to 50+ processes and hit context-switching overhead. At that point, a proper message broker wins. But for 2-8 workers on a single box doing CPU-heavy work? multiprocessing.Queue is still my go-to.

FAQ

Can I use multiprocessing.Queue with async FastAPI endpoints?
Yes, but don’t call .get() directly in an async route - it blocks the event loop. Use a separate worker process or run the Queue call in a threadpool via run_in_executor.

Is multiprocessing.Queue safe for multiple producers and consumers?
Yes. The Queue class handles locking internally so multiple processes can safely call .put() and .get() concurrently.

What’s the maximum size of a multiprocessing.Queue?
By default, it’s unlimited unless you specify a maxsize. If you set maxsize, .put() will block when the queue is full unless you use a timeout.

How do I monitor queue size in production?
You can’t rely on .qsize() for accuracy due to timing, but you can approximate it for logging. For real metrics, wrap your Queue in a proxy class that increments/decrements counters on put/get.

Key Takeaways

multiprocessing.Queue is a reliable, built-in python message queue between processes for FastAPI workers
Use it for CPU-bound tasks where you don’t need persistence or network transparency
Serialize carefully - prefer primitives over complex objects
Avoid .empty() and .qsize() for control flow; use timeouts and sentinel values instead
Benchmark your specific workload, but prioritize correctness over raw speed
Link to related production tips: FastAPI Async vs Sync: Benchmarks and When to Use Each and Bolt.new Python Backend: FastAPI from Zero to Production