В чем преимущество Celery перед ручной реализацией очереди задач?

# Celery vs ручная реализация очереди задач

Краткий ответ

Celery — это production-ready task queue для Python, которая:

• Гарантирует доставку задач
• Управляет workers'ами
• Обрабатывает ошибки и retry
• Масштабируется на тысячи workers

Если писать очередь руками, потребуется месяцы разработки и багов. Celery это делает за недели.

Ручная реализация очереди (наивный подход)

import queue
import threading
import time

# Простая очередь в памяти
task_queue = queue.Queue()

def simple_worker():
    while True:
        try:
            task = task_queue.get(timeout=1)
            print(f"Выполняю задачу: {task['name']}")
            time.sleep(task['duration'])
            print(f"Задача {task['name']} завершена")
            task_queue.task_done()
        except queue.Empty:
            pass

# Добавление задач
task_queue.put({'name': 'send_email', 'duration': 2})
task_queue.put({'name': 'generate_report', 'duration': 5})

# Запуск worker'ов
for _ in range(3):
    threading.Thread(target=simple_worker, daemon=True).start()

time.sleep(10)

Проблемы ручной реализации

1. Нет персистентности — если сервер перезагружается, задачи теряются
2. Нет масштабирования — worker'ы привязаны к одному процессу/машине
3. Нет retry механизма — если задача упала, она просто удалилась
4. Нет мониторинга — не видишь, какие задачи выполняются, сколько workers
5. Нет приоритизации — все задачи выполняются в одном порядке
6. Нет scheduled задач — нельзя запланировать выполнение на время

Celery реализация (production-ready)

from celery import Celery
from celery.schedules import crontab
import time

# 1. Создание Celery приложения
app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',      # Очередь: Redis
    backend='redis://localhost:6379/1'      # Результаты: Redis
)

# 2. Определение задач
@app.task(bind=True, max_retries=3)
def send_email(self, email, subject):
    """Отправить email с retry механизмом"""
    try:
        print(f"Отправляю email на {email}")
        time.sleep(2)
        return f"Email отправлен на {email}"
    except Exception as exc:
        # Retry через 5 секунд (exponential backoff)
        raise self.retry(exc=exc, countdown=5, max_retries=3)

@app.task
def generate_report(report_id):
    """Генерировать отчёт"""
    print(f"Генерирую отчёт {report_id}")
    time.sleep(5)
    return f"Отчёт {report_id} готов"

# 3. Задачи с расписанием
@app.task
def cleanup_old_files():
    """Очистить старые файлы каждый день в 2 часа ночи"""
    print("Очищаю старые файлы")

# Beat Scheduler для расписания
app.conf.beat_schedule = {
    'cleanup-every-day': {
        'task': 'myapp.cleanup_old_files',
        'schedule': crontab(hour=2, minute=0),  # 2:00 AM каждый день
    },
    'send-daily-digest': {
        'task': 'myapp.send_daily_digest',
        'schedule': 3600.0,  # каждый час
    },
}

# 4. Использование (в Django view или где угодно)
def my_view():
    # Асинхронно отправить email
    send_email.delay('user@example.com', 'Welcome!')
    
    # С параметрами и задержкой
    send_email.apply_async(
        args=('user@example.com', 'Hello!'),
        countdown=60  # Выполнить через 60 секунд
    )
    
    # Составная задача
    from celery import chain, group, chord
    
    workflow = chain(
        send_email.s('user@example.com', 'Part 1'),
        generate_report.s(123),
        send_email.s('user@example.com', 'Part 2')
    )
    result = workflow.apply_async()
    
    # Ждать результата
    print(result.get(timeout=30))

Преимущества Celery

1. Персистентность (не теряет задачи)

# Ручная реализация
task_queue = queue.Queue()  # В памяти! Теряется при перезагрузке

# Celery
send_email.delay('user@example.com')  # Сохраняется в Redis/RabbitMQ
# Даже если worker упадёт, задача ждёт в очереди

2. Масштабирование на распределённую систему

# Запусти worker'ов на разных машинах
celery -A myapp worker --loglevel=info --concurrency=4

# На машине 1: celery -A myapp worker --queues=emails
# На машине 2: celery -A myapp worker --queues=reports
# На машине 3: celery -A myapp worker --queues=images

# Celery распределит задачи по ним

3. Retry механизм

@app.task(bind=True, max_retries=3)
def unreliable_api_call(self, url):
    try:
        response = requests.get(url, timeout=5)
        return response.json()
    except requests.RequestException as exc:
        # Retry: 5s, 25s, 125s (exponential backoff)
        raise self.retry(exc=exc, countdown=5**self.request.retries)

# Ручная реализация требует своего retry механизма

4. Мониторинг и управление

# Flower — web UI для мониторинга Celery
# pip install flower
# flower -A myapp

# Показывает:
# - Активные задачи
# - Историю выполнения
# - Статус workers
# - Статистику
# - Очереди

# Из Python:
from celery.app.control import Inspect

app = Celery()
insp = Inspect(app=app)
print(insp.active())        # Активные задачи
print(insp.registered())    # Зарегистрированные задачи
print(insp.stats())         # Статистика workers

5. Приоритизация

# Задачи с приоритетом
send_email.apply_async(
    args=('vip@example.com',),
    priority=10  # Выше приоритет = раньше выполнится
)

send_email.apply_async(
    args=('regular@example.com',),
    priority=1
)

6. Scheduled задачи (Celery Beat)

# Ручная реализация
import schedule
import time

schedule.every().day.at("10:30").do(job)
schedule.every().hour.do(another_job)

while True:
    schedule.run_pending()
    time.sleep(60)

# Celery (проще и надёжнее)
app.conf.beat_schedule = {
    'every-morning': {
        'task': 'myapp.send_morning_digest',
        'schedule': crontab(hour=9, minute=0),
    },
}

# celery -A myapp beat

7. Комплексные workflows

from celery import chain, group, chord

# Последовательное выполнение
workflow = chain(
    task1.s(),
    task2.s(),
    task3.s()
)

# Параллельное выполнение
parallel = group(
    task1.s(),
    task2.s(),
    task3.s()
)

# Callback после параллельных
callback_workflow = chord([
    task1.s(),
    task2.s(),
    task3.s()
])(summarize.s())  # Выполнится после всех трёх

result = workflow.apply_async()
result.get()

8. Обработка результатов

# Ручная реализация
results = []
for task in tasks:
    if task.completed:
        results.append(task.result)

# Celery
result = send_email.delay('user@example.com')
print(result.id)              # ID задачи
print(result.status)          # PENDING, STARTED, SUCCESS, FAILURE
print(result.ready())         # Завершена ли?
print(result.get(timeout=10)) # Ждать результат
print(result.info)            # Информация о задаче

Сравнение: матрица

Реальный пример: миграция с ручной на Celery

# ДО: Ручная реализация
class SendEmailWorker(Thread):
    def run(self):
        while True:
            email_task = db.query(EmailTask).filter(
                status='pending'
            ).first()
            
            if email_task:
                try:
                    send_smtp(email_task.email)
                    db.update(email_task, status='completed')
                except Exception as e:
                    email_task.retry_count += 1
                    if email_task.retry_count < 3:
                        db.update(email_task, status='pending')
                    else:
                        db.update(email_task, status='failed')
            time.sleep(5)

# ПОСЛЕ: Celery
@app.task(bind=True, max_retries=3)
def send_email_task(self, email, subject):
    try:
        send_smtp(email)
        return f"Sent to {email}"
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

# И используется везде
send_email_task.delay('user@example.com', 'Hello')

Когда НЕ нужна Celery

1. Очень быстрые задачи (< 100ms) — overhead Celery замедлит
2. Простые скрипты (один раз в день) — schedule в крон
3. Внутри одного процесса — используй threading
4. Прототип без распределённости — сначала ручная, потом Celery

Вывод

Ручная реализация очереди потребует:

• Недель разработки
• Месяцев тестирования
• Постоянной поддержки и багов

Celery даёт:

• Production-ready решение за день
• Масштабирование на тысячи worker'ов
• Мониторинг и управление
• Retry, scheduling, workflows
• Надёжность, которую разработчики искали годами

Правило большого пальца: Если нужна асинхронная очередь задач и более чем 1 worker, используй Celery. Иначе стоимость ручной разработки перекроет затраты на изучение Celery.