Redis PubSub with Multithreading: Solving the background processing problem

The need for background processing is growing rapidly with AI use cases, especially when synchronous operations are untenable. AI document processing, image and video generation, deep research agents, and complex automations all share a common requirement: users submit requests and return later for results while orchestrations run on the backend, isolated from user-facing threads. The ideal architecture separates concerns into distinct microservices: one service handles user requests, another processes jobs. Connect them via Kafka, write metadata to a database, and scale independently. This setup works beautifully - if you have the infrastructure budget. The practical approach for indie hackers and MVPs is simpler: a single FastAPI service and a Redis instance. It's cost effective and sufficient for early-stage applications, but there's a critical gotcha: thread blocking.

Here is a typical flow for background job processing:

• User submits a document
• Save file to S3, metadata to MongoDB
• Publish message to Redis PubSub to trigger processing
• Handler processes the document in the background

Here is the gotcha. When Redis handles a message, it uses the main thread by default. Without special handling, you get main thread blocking and an unresponsive API.

# ❌ This blocks your main thread
def message_handler(message):
# Process document batch - takes 10+ minutes
process_batch(message["data"])

redis_client.subscribe(channel, message_handler)

# When setting up the subscription on FastAPI
@app.on_event("startup")
async def start_app():
# This subscription blocks the main thread during message processing
redis.subscribe(RedisChannels.DOCUMENT_INGESTION, ingestion.process_batch)
# API becomes unresponsive during batch processing

The issue is that Redis processes messages synchronously on the thread that established the subscription and in a FastAPI application, this is typically your main event loop thread, causing complete application blocking. My solution combines Python's `ThreadPoolExecutor` with isolated async execution contexts to achieve non-blocking message processing. This architecture bridges the gap between async/await paradigm and the need for CPU bound parallelism.

Understanding the solution

The key insight is that Python's asyncio provides concurrency, but THreadPoolExecutor provides parallelism. By combining both, we have:

1. Non-blocking I/O: Async functions handle database queries, API calls, and file operations efficiently
2. Parallelism: Multiple messages process simultaneously on different CPU cores
3. Event loop isolation: Each async handler runs in the main event loop without blocking it.

This' how the components work together:

This pattern works for the following reasons:

• Problem 1: Redis client blocks on `pubsub.listen()` waiting for messages. If run on the main thread, all HTTP requests hang and the API becomes unresponsive. The application appears frozen to users. The solution is to run the listener in a dedicated daemon thread, keeping the main thread free to handle HTTP requests
• Problem 2: Message handlers block the listener. Processing messages in the listener thread creates a sequential bottleness. Messages are processed one at a time so that n messages take 10n minutes if processing a single message takes 10 minutes. The solution is to submit each message to the ThreadPoolExecutor for parallel processing, allowing mutiple messages to process simultaneously.
• Problem 3: Async Handlers Can't Run Directly in Worker Threads. Worker threads don't have their own event loops and we can't use asynchronous calls in regular thread functions. So, async/await fail with 'no-running-event-loop' errors. You lose access to async database operations, HTTP clients and all async resources. The solution is to use asyncio.run_coroutine_threadsafe() to execute async code in mian event loop from worker thread context, maintaining access ot all async resources
• Problem 4: FastAPI needs responsive event loop. Long running tasks in the main loop block HTTP request handling. During document processing (which takes 2 min for example), all API endpoints return timeouts. Users see 504 errors and the service appears down. The solution is to have all heady processing happen in worker threads, main loop only schedules coroutines and remains free to handle incoming HTTP requests instantly.