Что такое Celery

Celery — это распределённая система очередей задач для Python. Она позволяет выполнять задачи асинхронно, планировать выполнение и обрабатывать ошибки.

▸Компоненты Celery

Broker — посредник для передачи задач (Redis, RabbitMQ)

Worker — процесс, выполняющий задачи

Result Backend — хранилище результатов (Redis, PostgreSQL)

Настройка

▸Установка

bash

1pip install celery[redis]

▸Конфигурация (celery.py)

python

1from celery import Celery
2
3app = Celery('myproject')
4
5app.config_from_object('django.conf:settings', namespace='CELERY')
6
7# Автоматическое обнаружение задач
8app.autodiscover_tasks()
9
10# Конфигурация (settings.py)
11CELERY_BROKER_URL = 'redis://localhost:6379/0'
12CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
13CELERY_ACCEPT_CONTENT = ['json']
14CELERY_TASK_SERIALIZER = 'json'
15CELERY_RESULT_SERIALIZER = 'json'

Определение задач

▸Простые задачи

python

1# tasks.py
2from celery import shared_task
3import time
4
5@shared_task
6def add(x, y):
7    return x + y
8
9@shared_task
10def process_data(data_id):
11    data = Data.objects.get(id=data_id)
12    # Длительная обработка
13    result = heavy_computation(data)
14    data.result = result
15    data.save()
16    return result

▸Задачи с результатами

python

1@shared_task(bind=True, max_retries=3)
2def send_email(self, to, subject, body):
3    try:
4        # Отправка email
5        smtp_send(to, subject, body)
6    except Exception as exc:
7        # Повторная попытка через 60 секунд
8        self.retry(exc=exc, countdown=60)

▸Задачи с таймаутами

python

1@shared_task(time_limit=300, soft_time_limit=250)
2def long_running_task():
3    # 5 минут максимум, предупреждение через 4 минуты 10 секунд
4    time.sleep(300)

Вызов задач

▸Асинхронный вызов

python

1# Из Django view
2from .tasks import process_data
3
4def my_view(request):
5    data_id = request.POST['data_id']
6    # Задача будет выполнена асинхронно
7    task = process_data.delay(data_id)
8    return JsonResponse({'task_id': task.id})

▸Ожидание результата

python

1# Синхронное ожидание (не рекомендуется в web)
2result = process_data.delay(data_id).get(timeout=10)
3
4# Асинхронное ожидание
5from celery.result import AsyncResult
6
7task = process_data.delay(data_id)
8# Позже проверяем результат
9result = AsyncResult(task.id)
10if result.ready():
11    data = result.get()

Планировщик (Celery Beat)

python

1# settings.py
2CELERY_BEAT_SCHEDULE = {
3    'cleanup-every-hour': {
4        'task': 'myapp.tasks.cleanup',
5        'schedule': crontab(minute=0),  # Каждый час
6    },
7    'daily-report': {
8        'task': 'myapp.tasks.generate_report',
9        'schedule': crontab(hour=8, minute=0),  # Каждый день в 8:00
10    },
11    'weekly-cleanup': {
12        'task': 'myapp.tasks.weekly_cleanup',
13        'schedule': crontab(hour=2, minute=0, day_of_week=0),  # Каждое воскресенье в 2:00
14    },
15}

Группы и цепочки

python

1from celery import group, chain, chord
2
3# Группа — параллельное выполнение
4job = group(add.s(i, i) for i in range(10))
5result = job.apply_async()
6
7# Цепочка — последовательное выполнение
8job = chain(
9    fetch_data.s(url),
10    process_data.s(),
11    save_result.s()
12)
13result = job.apply_async()
14
15# Аккорд — цепочка с колбэком
16header = group(fetch_data.s(url) for url in urls)
17callback = process_results.s()
18result = chord(header)(callback)

Мониторинг

▸Flower

bash

1pip install flower
2celery -A myproject flower

▸Просмотр задач

python

1from celery.result import AsyncResult
2
3task_id = 'abc123'
4result = AsyncResult(task_id)
5
6print(result.state)  # PENDING, STARTED, SUCCESS, FAILURE
7print(result.info)   # Информация о задаче
8print(result.result) # Результат (если есть)

Обработка ошибок

python

1from celery.exceptions import MaxRetriesExceededError
2
3@shared_task(bind=True, max_retries=3)
4def reliable_task(self):
5    try:
6        # Основная логика
7        do_something()
8    except ConnectionError as exc:
9        # Повторная попытка
10        raise self.retry(exc=exc, countdown=60)
11    except MaxRetriesExceededError:
12        # Логика после исчерпания попыток
13        notify_admin("Task failed after 3 retries")
14    except Exception as exc:
15        # Логирование неожиданных ошибок
16        logger.error(f"Unexpected error: {exc}")
17        raise

Best Practices

▸Идемпотентность

python

1@shared_task
2def process_order(order_id):
3    order = Order.objects.get(id=order_id)
4    if order.status == 'processed':
5        return  # Уже обработан
6    # Обработка...
7    order.status = 'processed'
8    order.save()

▸Размер задач

code

1# Плохо: большая задача
2@shared_task
3def process_all_users():
4    for user in User.objects.all():
5        process_user(user)
6
7# Хорошо: маленькие задачи
8@shared_task
9def process_user(user_id):
10    user = User.objects.get(id=user_id)
11    # Обработка одного пользователя

Заключение

Celery — стандарт для фоновых задач в Python-приложениях. Понимание broker, workers, планировщика и паттернов использования критически важно для масштабируемых приложений. На собеседовании спрашивают про идемпотентность, обработку ошибок и мониторинг задач.

#python#celery#task-queue#redis#async

Python≈ 14 мин чтения23 марта 2024 г.

Celery: фоновые задачи и очереди

Полное руководство по Celery: настройка, задачи, планировщик, мониторинг, лучшие практики.

Что такое Celery

▸Компоненты Celery

Broker — посредник для передачи задач (Redis, RabbitMQ)

Worker — процесс, выполняющий задачи

Result Backend — хранилище результатов (Redis, PostgreSQL)

Настройка

▸Установка

bash

1pip install celery[redis]

▸Конфигурация (celery.py)

python

1from celery import Celery
2
3app = Celery('myproject')
4
5app.config_from_object('django.conf:settings', namespace='CELERY')
6
7# Автоматическое обнаружение задач
8app.autodiscover_tasks()
9
10# Конфигурация (settings.py)
11CELERY_BROKER_URL = 'redis://localhost:6379/0'
12CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
13CELERY_ACCEPT_CONTENT = ['json']
14CELERY_TASK_SERIALIZER = 'json'
15CELERY_RESULT_SERIALIZER = 'json'

Определение задач

▸Простые задачи

python

1# tasks.py
2from celery import shared_task
3import time
4
5@shared_task
6def add(x, y):
7    return x + y
8
9@shared_task
10def process_data(data_id):
11    data = Data.objects.get(id=data_id)
12    # Длительная обработка
13    result = heavy_computation(data)
14    data.result = result
15    data.save()
16    return result

▸Задачи с результатами

python

1@shared_task(bind=True, max_retries=3)
2def send_email(self, to, subject, body):
3    try:
4        # Отправка email
5        smtp_send(to, subject, body)
6    except Exception as exc:
7        # Повторная попытка через 60 секунд
8        self.retry(exc=exc, countdown=60)

▸Задачи с таймаутами

python

1@shared_task(time_limit=300, soft_time_limit=250)
2def long_running_task():
3    # 5 минут максимум, предупреждение через 4 минуты 10 секунд
4    time.sleep(300)

Вызов задач

▸Асинхронный вызов

python

1# Из Django view
2from .tasks import process_data
3
4def my_view(request):
5    data_id = request.POST['data_id']
6    # Задача будет выполнена асинхронно
7    task = process_data.delay(data_id)
8    return JsonResponse({'task_id': task.id})

▸Ожидание результата

python

1# Синхронное ожидание (не рекомендуется в web)
2result = process_data.delay(data_id).get(timeout=10)
3
4# Асинхронное ожидание
5from celery.result import AsyncResult
6
7task = process_data.delay(data_id)
8# Позже проверяем результат
9result = AsyncResult(task.id)
10if result.ready():
11    data = result.get()

Планировщик (Celery Beat)

python

1# settings.py
2CELERY_BEAT_SCHEDULE = {
3    'cleanup-every-hour': {
4        'task': 'myapp.tasks.cleanup',
5        'schedule': crontab(minute=0),  # Каждый час
6    },
7    'daily-report': {
8        'task': 'myapp.tasks.generate_report',
9        'schedule': crontab(hour=8, minute=0),  # Каждый день в 8:00
10    },
11    'weekly-cleanup': {
12        'task': 'myapp.tasks.weekly_cleanup',
13        'schedule': crontab(hour=2, minute=0, day_of_week=0),  # Каждое воскресенье в 2:00
14    },
15}

Группы и цепочки

python

1from celery import group, chain, chord
2
3# Группа — параллельное выполнение
4job = group(add.s(i, i) for i in range(10))
5result = job.apply_async()
6
7# Цепочка — последовательное выполнение
8job = chain(
9    fetch_data.s(url),
10    process_data.s(),
11    save_result.s()
12)
13result = job.apply_async()
14
15# Аккорд — цепочка с колбэком
16header = group(fetch_data.s(url) for url in urls)
17callback = process_results.s()
18result = chord(header)(callback)

Мониторинг

▸Flower

bash

1pip install flower
2celery -A myproject flower

▸Просмотр задач

python

1from celery.result import AsyncResult
2
3task_id = 'abc123'
4result = AsyncResult(task_id)
5
6print(result.state)  # PENDING, STARTED, SUCCESS, FAILURE
7print(result.info)   # Информация о задаче
8print(result.result) # Результат (если есть)

Обработка ошибок

python

1from celery.exceptions import MaxRetriesExceededError
2
3@shared_task(bind=True, max_retries=3)
4def reliable_task(self):
5    try:
6        # Основная логика
7        do_something()
8    except ConnectionError as exc:
9        # Повторная попытка
10        raise self.retry(exc=exc, countdown=60)
11    except MaxRetriesExceededError:
12        # Логика после исчерпания попыток
13        notify_admin("Task failed after 3 retries")
14    except Exception as exc:
15        # Логирование неожиданных ошибок
16        logger.error(f"Unexpected error: {exc}")
17        raise

Best Practices

▸Идемпотентность

python

1@shared_task
2def process_order(order_id):
3    order = Order.objects.get(id=order_id)
4    if order.status == 'processed':
5        return  # Уже обработан
6    # Обработка...
7    order.status = 'processed'
8    order.save()

▸Размер задач

code

1# Плохо: большая задача
2@shared_task
3def process_all_users():
4    for user in User.objects.all():
5        process_user(user)
6
7# Хорошо: маленькие задачи
8@shared_task
9def process_user(user_id):
10    user = User.objects.get(id=user_id)
11    # Обработка одного пользователя

Заключение

#python#celery#task-queue#redis#async

🛠

Попробуйте наш инструмент

Закрепите знания с помощью интерактивных упражнений

Перейти →

Что такое Celery

▸Компоненты Celery

Настройка

▸Установка

▸Конфигурация (celery.py)

Определение задач

▸Простые задачи

▸Задачи с результатами

▸Задачи с таймаутами

Вызов задач

▸Асинхронный вызов

▸Ожидание результата

Планировщик (Celery Beat)

Группы и цепочки

Мониторинг

▸Flower

▸Просмотр задач

Обработка ошибок

Best Practices

▸Идемпотентность

▸Размер задач

Заключение

Попробуйте наш инструмент

Похожие статьи

React: полное руководство для собеседования

JavaScript: глубокий анализ для собеседования

Node.js и backend: что спрашивают на собеседовании

DevOps: от основ до продвинутого уровня

Что такое Celery

▸Компоненты Celery

Настройка

▸Установка

▸Конфигурация (celery.py)

Определение задач

▸Простые задачи

▸Задачи с результатами

▸Задачи с таймаутами

Вызов задач

▸Асинхронный вызов

▸Ожидание результата

Планировщик (Celery Beat)

Группы и цепочки

Мониторинг

▸Flower

▸Просмотр задач

Обработка ошибок

Best Practices

▸Идемпотентность

▸Размер задач

Заключение

Попробуйте наш инструмент

Похожие статьи

React: полное руководство для собеседования

JavaScript: глубокий анализ для собеседования

Node.js и backend: что спрашивают на собеседовании

DevOps: от основ до продвинутого уровня