如何配置Celery实现按用户隔离的同步分布式任务处理并支持扩展？

Great question! Your current approach of binding single-concurrency workers to fixed queues gets the job done, but it's inflexible and doesn't scale well when task backlogs hit. Let's walk through how to adjust Celery to meet your needs, plus a few alternative tools if you want to explore other options.

Celery Solution: Distributed Locks + Dynamic Queue Routing #

Instead of tying workers to specific queues, we'll use distributed locks to ensure only one task runs per user queue at a time, while letting your 5 workers pull tasks from any queue. Here's how to set it up:

1. Route Tasks to Per-User Queues #

First, make sure each user's tasks are sent to their own dedicated queue (e.g., user_123 for user ID 123). When submitting tasks, specify the queue explicitly:

# In your task submission code
from app.celery import cpu_intensive_task

user_id = 123
task_data = {"foo": "bar"}
cpu_intensive_task.apply_async(
    args=[user_id, task_data],
    queue=f"user_{user_id}"
)

2. Add Distributed Locks to Tasks #

Use a lock backend (Redis works perfectly here, since you're likely already using it for Celery's broker/backend) to block concurrent execution for the same user queue. Here's a task example with a Redis lock:

import redis
from celery import Celery

# Initialize Celery and Redis client
celery = Celery("app", broker="redis://localhost:6379/0", backend="redis://localhost:6379/0")
redis_client = redis.Redis(host="localhost", port=6379, db=0)

@celery.task(bind=True)
def cpu_intensive_task(self, user_id, task_data):
    queue_lock_key = f"queue_lock:user_{user_id}"
    
    # Acquire lock with a timeout (prevents stale locks if tasks crash)
    with redis_client.lock(queue_lock_key, timeout=3600):
        # Execute your CPU-heavy work here
        result = process_cpu_intensive_work(task_data)
        return f"Task for user {user_id} completed: {result}"

3. Start Workers to Listen to All User Queues #

Launch your 5 workers to monitor all user-specific queues (using a wildcard pattern if your queues follow a consistent naming convention):

celery -A app.celery worker --concurrency=5 --queues=user_* --logfile=celery.log

Why This Works:

• Flexible Worker Utilization: Your 5 workers can pull tasks from any user queue, so no worker sits idle while another queue has backlogged tasks.
• Guaranteed Serial Execution: The lock ensures only one task runs per user queue at a time.
• Scalable: You can add new user queues dynamically without restarting workers.

Alternative Task Systems #

If you want to explore tools beyond Celery, these options also support per-queue serial execution with flexible worker routing:

• RQ (Redis Queue): A lightweight, Redis-based task queue. You can pair RQ's @job decorator with Redis locks to enforce serial per-user tasks. Workers can listen to multiple queues, making it easy to share resources across users.
• Dramatiq: A high-performance task queue with built-in support for task locking and priority queues. It's optimized for CPU-intensive workloads, and you can configure workers to consume from multiple queues while enforcing serial execution per queue.
• Apache Airflow: If your tasks have complex dependencies or you need scheduling capabilities, Airflow lets you create per-user DAGs that execute tasks serially. It's heavier than the other options, but ideal for orchestrating batch workflows.

内容的提问来源于stack exchange，提问作者James Foley