如何配置Celery限制单节点最多运行10个需30GiB磁盘的任务？

Absolutely feasible! Given your scenario—each task uses ~30GiB of disk space, and your nodes have 600GiB available—limiting each node to run a maximum of 10 such tasks is totally practical, and there are a couple of straightforward ways to set this up with Celery and Redis.

Approach 1: Dedicated Queue + Worker Concurrency Limit #

This is the most clean and maintainable method, as it isolates your disk-intensive tasks to a dedicated worker explicitly configured for the desired concurrency.

Step 1: Route Tasks to a Dedicated Queue #

First, configure Celery to send your disk-heavy task to a specific queue (e.g., disk_intensive_tasks). Add this to your Celery app configuration:

from celery import Celery

app = Celery('your_app_name', broker='redis://localhost:6379/0')

# Route the specific task to its dedicated queue
app.conf.task_routes = {
    'your_module.your_disk_heavy_task': {'queue': 'disk_intensive_tasks'}
}

Step 2: Start a Worker for the Queue with Concurrency Limit #

Launch a Celery worker that only listens to the disk_intensive_tasks queue, and set its concurrency to 10. This ensures the worker will never run more than 10 of these tasks at once:

celery -A your_app_name worker -Q disk_intensive_tasks --concurrency=10 --loglevel=info

Since 10 tasks × 30GiB = 300GiB, this leaves plenty of headroom on your 600GiB nodes.

Approach 2: Task-Level Semaphore (Per-Node Limit) #

If you prefer not to run a separate dedicated worker, you can use a semaphore (backed by Redis, since you’re already using it as your broker) to restrict concurrent runs of the task on each individual node.

Implementation Code #

from celery import Celery
from redis import Redis
import socket

app = Celery('your_app_name', broker='redis://localhost:6379/0')
redis_client = Redis(host='localhost', port=6379, db=0)

# Unique semaphore key per node and task (uses hostname to distinguish nodes)
SEMAPHORE_KEY = f"celery:semaphore:disk_heavy_task:{socket.gethostname()}"
MAX_CONCURRENT_TASKS = 10

@app.task
def your_disk_heavy_task():
    # Acquire the semaphore before running the task
    with redis_client.lock(
        SEMAPHORE_KEY,
        timeout=3600,  # Lock expires after 1 hour to prevent deadlocks
        blocking=True,
        blocking_timeout=None  # Wait indefinitely for the lock if needed
    ):
        # Execute your disk-intensive task logic here
        perform_heavy_disk_operation()

This ensures that only 10 instances of the task can run on a single node at any time, while other tasks (if any) can run on separate workers without interference.

Key Notes #

• For the dedicated queue approach: Make sure your worker has enough CPU/memory to handle 10 concurrent tasks in addition to disk resources.
• For the semaphore approach: Ensure Redis is highly available (since it’s storing the lock), but since you’re already using it as your broker, this should be covered.
• Both methods scale well across multiple nodes—each node can run its own dedicated worker (with concurrency 10) or use the per-node semaphore, so your total cluster capacity is number_of_nodes × 10.

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