Можно ли добавить autoscale в Celery?

Autoscale в Celery

Да, Celery полностью поддерживает autoscale (автоматическое масштабирование). Это позволяет динамически изменять количество рабочих потоков в зависимости от нагрузки.

Концепция Autoscale

Autoscale в Celery работает следующим образом:

• Задаются минимальное и максимальное количество одновременных задач (pool size)
• Celery автоматически добавляет рабочие процессы когда очередь растёт
• И убирает рабочие процессы когда нагрузка снижается

Способ 1: Запуск worker с флагом autoscale

Самый простой способ — использовать флаг --autoscale при запуске worker.

# Синтаксис: --autoscale=max,min
# min — минимальное количество worker процессов
# max — максимальное количество worker процессов

celery -A myapp worker --autoscale=10,3
# Минимум 3 процесса, максимум 10

celery -A myapp worker --autoscale=50,5
# Минимум 5 процессов, максимум 50 (для high traffic)

Способ 2: Конфигурация через settings

Можно задать autoscale в конфигурации Celery.

# celery.py или settings.py
from celery import Celery

app = Celery('myapp')

# Конфигурация
app.conf.update(
    # Autoscale: максимум 20 задач, минимум 2
    worker_autoscaler='celery.worker.autoscale:Autoscaler',
    worker_autoscale='20,2',
    
    # Или альтернативный синтаксис
    autoscale_settings={
        'max_concurrency': 20,
        'min_concurrency': 2,
    }
)

Способ 3: Задачи с приоритетом и pooling

Целевой пулинг — это более сложный подход с приоритизацией задач.

from celery import Celery, Task
from celery.concurrency import pool

app = Celery('myapp')

# Несколько очередей с разными настройками
app.conf.task_queues = (
    {
        'name': 'default',
        'exchange': 'tasks',
        'routing_key': 'task.default',
    },
    {
        'name': 'high_priority',
        'exchange': 'tasks',
        'routing_key': 'task.high',
    },
)

# Worker для high priority может иметь другие параметры
# celery -A myapp worker -Q high_priority --autoscale=30,10
# celery -A myapp worker -Q default --autoscale=10,2

@app.task(queue='default')
def slow_task():
    pass

@app.task(queue='high_priority')
def fast_task():
    pass

Параметры Autoscale

app.conf.update(
    # Основные параметры
    worker_autoscale='30,10',  # max,min
    
    # Скорость масштабирования
    # Время между проверками нагрузки (в секундах)
    worker_max_tasks_per_child=1000,
    
    # Таймаут для worker процесса (если зависает)
    task_soft_time_limit=300,  # 5 минут
    task_time_limit=600,  # 10 минут (hard limit)
    
    # Pool-specific настройки
    worker_pool='celery.concurrency.prefork:TaskPool',
    worker_prefetch_multiplier=4,  # Сколько задач загружать заранее
)

Пример: Полная конфигурация с autoscale

# celery.py
import os
from celery import Celery
from kombu import Exchange, Queue

app = Celery('myapp')

# Redis broker
app.conf.broker_url = 'redis://localhost:6379/0'
app.conf.result_backend = 'redis://localhost:6379/1'

# Queues
default_exchange = Exchange('tasks', type='direct')
default_queue = Queue('default', exchange=default_exchange, routing_key='default')

app.conf.task_queues = (
    Queue('default', exchange=default_exchange, routing_key='default'),
    Queue('high_priority', exchange=default_exchange, routing_key='high'),
    Queue('low_priority', exchange=default_exchange, routing_key='low'),
)

# Default routing
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'tasks'
app.conf.task_default_routing_key = 'default'

# Autoscale settings
app.conf.update(
    worker_autoscale='50,5',  # min 5, max 50 concurrent tasks
    worker_prefetch_multiplier=4,
    task_acks_late=True,  # Acknowledge после выполнения
    task_reject_on_worker_lost=True,  # Переоправить если worker умер
    
    # Таймауты
    task_soft_time_limit=300,
    task_time_limit=600,
    
    # Результаты
    result_expires=3600,  # Удалять результаты через час
    result_compression='gzip',
)

# Задачи
@app.task(bind=True, queue='default')
def process_data(self, data):
    try:
        # Долгая операция
        result = expensive_operation(data)
        return result
    except Exception as e:
        self.retry(exc=e, countdown=60, max_retries=3)

@app.task(bind=True, queue='high_priority')
def urgent_task(self, user_id):
    # Быстрая операция, высокий приоритет
    return send_notification(user_id)

Запуск workers с разными autoscale стратегиями

# Worker для обычных задач (aggressive scaling)
celery -A myapp worker -Q default \
  --autoscale=20,2 \
  --prefetch-multiplier=1 \
  --concurrency=4

# Worker для high priority задач (always ready)
celery -A myapp worker -Q high_priority \
  --autoscale=50,10 \
  --prefetch-multiplier=2 \
  --concurrency=8

# Worker для низкоприоритетных задач (conservative scaling)
celery -A myapp worker -Q low_priority \
  --autoscale=10,1 \
  --prefetch-multiplier=1 \
  --concurrency=2

Мониторинг autoscale

from celery import Celery
from celery.events.state import State
from celery.app.utils import Settings

app = Celery('myapp')

def print_worker_stats():
    inspect = app.control.inspect()
    
    # Получить статус workers
    active_tasks = inspect.active()
    stats = inspect.stats()
    reserved = inspect.reserved()
    
    for worker_name, tasks in (active_tasks or {}).items():
        print(f"{worker_name}: {len(tasks)} активных задач")
    
    for worker_name, worker_stats in (stats or {}).items():
        print(f"{worker_name}: pool={worker_stats['pool']}")
    
    for worker_name, reserved_tasks in (reserved or {}).items():
        print(f"{worker_name}: {reserved_tasks['reserved']} зарезервировано")

# Запустить
print_worker_stats()

Лучшие практики Autoscale

# ❌ Плохо: слишком агрессивное масштабирование
worker_autoscale='1000,1'  # Создаст тысячу процессов!

# ✅ Хорошо: разумные пределы
worker_autoscale='50,5'  # max 50, min 5

# ❌ Плохо: очень высокое prefetch
worker_prefetch_multiplier=100

# ✅ Хорошо: низкое prefetch для быстрого масштабирования
worker_prefetch_multiplier=1  # или 2

# ❌ Плохо: без таймаутов
app.conf.update()  # Никаких таймаутов

# ✅ Хорошо: установлены таймауты
app.conf.update(
    task_soft_time_limit=300,   # 5 минут
    task_time_limit=600,        # 10 минут
)

# ❌ Плохо: ack до выполнения
app.conf.task_acks_late = False

# ✅ Хорошо: ack после выполнения
app.conf.task_acks_late = True

Интеграция с Docker/Kubernetes

# Dockerfile
FROM python:3.11

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .

# Запуск worker с autoscale
CMD ["celery", "-A", "myapp", "worker", \
     "--autoscale=50,5", \
     "--loglevel=info", \
     "--concurrency=4"]

# kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
spec:
  replicas: 3  # Несколько worker pod'ов
  template:
    spec:
      containers:
      - name: celery
        image: myapp:latest
        env:
        - name: CELERY_BROKER_URL
          value: "redis://redis:6379/0"
        resources:
          requests:
            memory: "256Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"

Альтернативы: Horizontal Scaling

Частый подход — вместо autoscale, использовать горизонтальное масштабирование.

# Вместо одного worker с autoscale=50,5
celery -A myapp worker --autoscale=50,5

# Запустить несколько workers (например, в Kubernetes)
kubectl scale deployment celery-worker --replicas=10

# Каждый worker может иметь фиксированный размер
celery -A myapp worker --concurrency=4

Этот подход часто проще управлять в контейнеризованной среде.

Итог

Да, Celery поддерживает autoscale полностью:

1. Встроенная поддержка через флаг --autoscale=max,min
2. Гибкая конфигурация через settings
3. Поддержка очередей с разными стратегиями масштабирования
4. Мониторинг и контроль через inspect API

Для production рекомендую:

• Использовать Kubernetes с горизонтальным масштабированием
• Или Autoscale для простых случаев
• Всегда устанавливать таймауты
• Тестировать поведение под нагрузкой