Как работает Celery на Python руководство для разработчиков

Celery – это мощный инструмент для выполнения асинхронных задач в Python. Если вы хотите разгрузить основной поток вашего приложения, начните с установки Celery через pip install celery. После этого создайте файл tasks.py, где определите функции, которые будут выполняться в фоновом режиме. Для запуска Celery используйте команду celery -A tasks worker —loglevel=info.

Celery работает через брокеры сообщений, такие как RabbitMQ или Redis. Эти брокеры позволяют передавать задачи между вашим приложением и воркерами Celery. Например, для настройки Redis в качестве брокера добавьте в ваш проект строку app = Celery(‘tasks’, broker=’redis://localhost:6379/0′). Это обеспечит надежную передачу задач и их выполнение в отдельном процессе.

Для выполнения задач в Celery используйте декоратор @app.task. Например, функция, которая отправляет электронные письма, может быть оформлена так: @app.task def send_email(to, subject, body): …. После этого вызовите эту функцию из вашего приложения с помощью send_email.delay(to, subject, body). Celery автоматически добавит задачу в очередь и выполнит ее асинхронно.

Celery поддерживает планирование задач через периодические задания. Для этого настройте celerybeat и добавьте задачи в файл конфигурации. Например, чтобы отправлять отчеты каждый день, используйте app.conf.beat_schedule = {‘send-reports’: {‘task’: ‘tasks.send_report’, ‘schedule’: crontab(hour=8)}}. Это позволяет автоматизировать рутинные процессы без дополнительных усилий.

Для мониторинга и управления задачами Celery предоставляет инструменты, такие как Flower. Установите его через pip install flower и запустите с помощью flower -A tasks —port=5555. Flower предоставляет веб-интерфейс, где вы можете отслеживать статус задач, их выполнение и ошибки.

Основы работы Celery с задачами и очередями

Создавайте задачи в Celery с помощью декоратора @app.task. Это позволяет превратить любую функцию в асинхронную задачу, которая может выполняться в фоновом режиме. Например:

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y

Для запуска задачи используйте метод delay. Он помещает задачу в очередь для выполнения:

result = add.delay(4, 6)

Celery поддерживает несколько брокеров сообщений, таких как Redis, RabbitMQ и Amazon SQS. Укажите брокер в настройках приложения:

app.conf.broker_url = 'redis://localhost:6379/0'

Очереди в Celery позволяют распределять задачи между воркерами. По умолчанию задачи отправляются в очередь celery. Чтобы указать другую очередь, используйте параметр queue:

add.apply_async((4, 6), queue='high_priority')

Настройте маршрутизацию задач, чтобы автоматически распределять их по очередям. Добавьте в конфигурацию:

app.conf.task_routes = {
'tasks.add': {'queue': 'high_priority'},
}

Для запуска воркеров, которые будут обрабатывать задачи, используйте команду:

celery -A tasks worker --loglevel=info

Управляйте состоянием задач с помощью методов ready, get и status. Например:

if result.ready():
print(result.get())

Для отладки и мониторинга используйте Flower – инструмент для визуализации состояния воркеров и задач. Установите его и запустите:

pip install flower
celery -A tasks flower

Celery поддерживает повторное выполнение задач при сбоях. Настройте параметры повторных попыток в задаче:

@app.task(bind=True, max_retries=3)
def retry_task(self):
try:
# Логика задачи
except Exception as e:
raise self.retry(exc=e, countdown=5)

Используйте периодические задачи с Celery Beat. Настройте расписание в конфигурации:

from celery.schedules import crontab
app.conf.beat_schedule = {
'add-every-minute': {
'task': 'tasks.add',
'schedule': crontab(minute='*/1'),
'args': (16, 16),
},
}

Эти основы помогут вам эффективно использовать Celery для управления задачами и очередями в вашем проекте.

Что такое задачи в Celery и как их определять

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y

После определения задачи вы можете вызывать её с помощью метода delay, который помещает задачу в очередь для выполнения. Например:

result = add.delay(4, 6)

Используйте метод get для получения результата задачи. Учтите, что вызов get блокирует выполнение до завершения задачи:

print(result.get())

Для задач, которые не требуют возврата результата, используйте параметр ignore_result=True. Это уменьшит нагрузку на брокер сообщений:

@app.task(ignore_result=True)
def log_message(message):
print(message)

Если задача может выполняться долго, установите тайм-аут с помощью параметра soft_time_limit или time_limit. Это поможет избежать зависаний:

@app.task(soft_time_limit=10, time_limit=15)
def process_data(data):
# Обработка данных
pass

Для работы с повторяющимися задачами используйте Celery Beat. Определите расписание в конфигурации Celery или через декоратор @app.on_after_configure.connect. Например:

from celery.schedules import crontab
app.conf.beat_schedule = {
'every-minute-task': {
'task': 'tasks.add',
'schedule': crontab(minute='*/1'),
'args': (2, 2),
},
}

Следите за состоянием задач с помощью методов ready, successful и failed. Это поможет контролировать выполнение и обрабатывать ошибки:

if result.ready():
if result.successful():
print("Задача выполнена успешно")
else:
print("Задача завершилась с ошибкой")

Определяйте задачи с учётом их назначения и требований к производительности. Это обеспечит стабильную работу вашего приложения.

Как настроить брокер сообщений для работы с Celery

Для начала установите выбранный брокер сообщений. Например, для RabbitMQ используйте команду:

sudo apt-get install rabbitmq-server

После установки запустите RabbitMQ и добавьте пользователя:

sudo systemctl start rabbitmq-server
sudo rabbitmqctl add_user myuser mypassword
sudo rabbitmqctl set_user_tags myuser administrator
sudo rabbitmqctl set_permissions -p / myuser ".*" ".*" ".*"

Для Redis установите его через пакетный менеджер:

sudo apt-get install redis-server

Затем проверьте, что Redis запущен:

sudo systemctl status redis

В проекте с Celery укажите брокер в настройках. Для RabbitMQ добавьте в celery.py:

app = Celery('myapp', broker='amqp://myuser:mypassword@localhost:5672//')

Для Redis используйте:

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

Проверьте подключение, запустив Celery:

celery -A myapp worker --loglevel=info

Если все настроено правильно, Celery подключится к брокеру и начнет обработку задач.

Примеры создания и вызова задач

Создайте задачу в Celery, используя декоратор @app.task. Например, функция для сложения двух чисел может выглядеть так:

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y

Для вызова задачи используйте метод delay. Это отправляет задачу в очередь для выполнения:

result = add.delay(4, 6)

Если нужно получить результат выполнения задачи, вызовите метод get:

print(result.get())

Для асинхронного выполнения задач с параметрами передайте их в delay:

result = add.delay(10, 20)

Чтобы создать задачу с аргументами по умолчанию, просто добавьте их в функцию:

@app.task
def multiply(x, y=2):
return x * y
result = multiply.delay(5)  # Умножит 5 на 2

Если задача требует выполнения через определенное время, используйте метод apply_async с параметром countdown:

result = add.apply_async((3, 7), countdown=10)  # Задача выполнится через 10 секунд

Для задач, которые должны выполняться периодически, настройте расписание в Celery Beat. Добавьте задачу в конфигурацию:

app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16),
},
}

Используйте Celery для работы с длительными операциями, такими как обработка файлов или отправка электронных писем. Например:

@app.task
def send_email(to, subject, body):
# Логика отправки письма
pass
send_email.delay('user@example.com', 'Привет', 'Это тестовое письмо.')

Celery также поддерживает цепочки задач. Создайте последовательность с помощью chain:

from celery import chain
chain(add.s(2, 2), multiply.s(4)).apply_async()

Используйте группы для параллельного выполнения задач:

from celery import group
group(add.s(1, 1), add.s(2, 2), add.s(3, 3)).apply_async()

Эти примеры помогут вам быстро начать работу с Celery и интегрировать его в ваши проекты.

Расширенные возможности Celery для обработки задач

Используйте цепочки задач (chains) для выполнения последовательных операций. Например, если вам нужно обработать данные, а затем отправить результат, создайте цепочку с помощью chain(task1.s(), task2.s()). Это упрощает управление зависимостями между задачами.

Группы задач (groups) позволяют выполнять несколько независимых операций параллельно. Создайте группу с помощью group(task1.s(), task2.s()), чтобы запустить задачи одновременно. Это полезно для обработки большого объема данных или выполнения множества запросов.

Используйте сигналы (signals) для отслеживания событий в Celery. Например, сигнал task_success позволяет выполнить код после успешного завершения задачи. Это помогает в мониторинге и логировании.

Настройте повторное выполнение задач с помощью параметра autoretry_for. Укажите исключения, при которых задача должна быть повторена, и задайте интервал между попытками. Это повышает надежность системы при временных сбоях.

Применяйте ограничения скорости (rate limits) для управления нагрузкой. Используйте параметр rate_limit, чтобы ограничить количество задач, выполняемых в единицу времени. Это полезно при работе с API, имеющими ограничения на количество запросов.

Для сложных сценариев используйте графы задач (chords). Графы позволяют объединить группу задач и выполнить финальную операцию после их завершения. Например, chord(group(task1.s(), task2.s()), task3.s()) выполнит task3 только после завершения task1 и task2.

Настройте планировщик (beat) для выполнения периодических задач. Создайте расписание в celery.py с помощью app.conf.beat_schedule. Это удобно для задач, которые должны выполняться регулярно, таких как отправка отчетов или обновление данных.

Используйте результаты задач для дальнейшей обработки. Сохраняйте результаты в базе данных или передавайте их между задачами с помощью AsyncResult. Это позволяет строить сложные рабочие процессы с минимальными усилиями.

Оптимизируйте производительность с помощью пула воркеров (worker pool). Выберите подходящий тип пула, например prefork для CPU-интенсивных задач или gevent для IO-операций. Это улучшит скорость выполнения задач.

Как применять планы выполнения задач (Periodic tasks)

Для настройки периодических задач в Celery используйте модуль celery.beat. Создайте объект celery.beat.Scheduler и определите задачи в формате crontab или interval. Например, чтобы запускать задачу каждые 10 минут, укажите:

from celery.schedules import crontab
app.conf.beat_schedule = {
'task-name': {
'task': 'module.task_function',
'schedule': crontab(minute='*/10'),
},
}

Для задач, которые должны выполняться с фиксированным интервалом, используйте timedelta:

from datetime import timedelta
app.conf.beat_schedule = {
'task-name': {
'task': 'module.task_function',
'schedule': timedelta(seconds=300),
},
}

Запустите планировщик задач командой celery -A your_app beat. Убедитесь, что Celery Worker также запущен для обработки задач.

Если требуется более гибкое управление, например, динамическое изменение расписания, используйте DatabaseScheduler. Он позволяет хранить задачи в базе данных и изменять их в реальном времени.

Для отладки периодических задач добавьте логирование в функции. Это поможет отслеживать выполнение и выявлять ошибки:

import logging
logger = logging.getLogger(__name__)
@app.task
def task_function():
logger.info('Task started')
# Логика задачи
logger.info('Task completed')

Проверяйте корректность расписания с помощью команды celery -A your_app inspect scheduled. Она покажет список запланированных задач и их параметры.

Управление зависимостями между задачами

Для управления зависимостями между задачами в Celery используйте цепочки (chains) и группы (groups). Цепочки позволяют последовательно выполнять задачи, где результат одной задачи передается в качестве аргумента следующей. Например, если вам нужно сначала обработать данные, а затем отправить их, создайте цепочку с помощью chain(task1.s(), task2.s()).

Группы полезны для параллельного выполнения задач. Вы можете запустить несколько независимых задач одновременно с помощью group(task1.s(), task2.s()). Это особенно удобно, когда задачи не зависят друг от друга, но их результаты нужны для дальнейшей обработки.

Для более сложных сценариев используйте сигнатуры (signatures) и холсты (canvas). Сигнатуры позволяют описывать задачи с их аргументами и настройками, а холсты объединяют их в сложные структуры. Например, с помощью chord можно сначала выполнить группу задач, а затем передать их результаты в финальную задачу.

Если задачи должны выполняться только при успешном завершении предыдущих, используйте chord или комбинируйте цепочки с группами. Это помогает избежать лишних вызовов и упрощает логику выполнения.

Для отладки и мониторинга зависимостей используйте встроенные инструменты Celery, такие как Flower. Они покажут, как задачи связаны между собой и в каком состоянии находятся.

Учитывайте, что ошибка в одной задаче может остановить выполнение всей цепочки. Чтобы избежать этого, добавьте обработку исключений в задачи или используйте link_error для выполнения задач в случае сбоя.

Мониторинг и отладка задач Celery

Используйте Flower для мониторинга задач Celery. Этот инструмент предоставляет веб-интерфейс, где можно отслеживать статус задач, просматривать их аргументы, время выполнения и ошибки. Установите Flower через pip и запустите его командой celery -A ваш_проект flower.

Для отладки задач настройте логирование. Добавьте в конфигурацию Celery параметры worker_hijack_root_logger и worker_log_format, чтобы логи были понятными и структурированными. Используйте модуль logging Python для записи дополнительной информации в лог-файлы.

Проверяйте состояние брокера сообщений, например, RabbitMQ или Redis. Убедитесь, что очередь задач не переполнена, а брокер работает стабильно. Для RabbitMQ используйте команду rabbitmqctl list_queues, а для Redis – redis-cli info.

Используйте celery inspect для проверки состояния воркеров. Команда celery -A ваш_проект inspect active покажет текущие выполняемые задачи, а celery -A ваш_проект inspect stats предоставит статистику по воркерам.

Для анализа ошибок задач используйте celery events. Этот инструмент позволяет отслеживать события в реальном времени, такие как начало выполнения задачи, её завершение или возникновение исключения. Запустите его командой celery -A ваш_проект events.

Если задача завершается с ошибкой, проверьте её трассировку стека. Celery сохраняет исключения в результатах задач, которые можно получить через task.result или в интерфейсе Flower. Это поможет быстро найти причину проблемы.

Для тестирования задач используйте celery.contrib.testing. Этот модуль позволяет запускать задачи в изолированной среде, что упрощает отладку без необходимости разворачивать полноценную инфраструктуру.

Регулярно обновляйте Celery и его зависимости. Новые версии часто содержат исправления ошибок и улучшения производительности, что может решить проблемы, с которыми вы сталкиваетесь.

Как конфигурировать Celery для масштабируемых приложений

Настройте брокер сообщений на использование RabbitMQ или Redis. RabbitMQ лучше подходит для высоконагруженных систем, а Redis – для простых сценариев с меньшими требованиями к производительности. Убедитесь, что брокер работает на отдельном сервере для повышения отказоустойчивости.

Используйте параметр worker_concurrency для управления количеством параллельных задач. Увеличьте его значение, если задачи выполняются быстро и не требуют много ресурсов. Например, для CPU-bound задач установите значение, равное количеству ядер процессора.

Для обработки большого числа задач добавьте несколько воркеров. Запустите их на разных серверах, чтобы распределить нагрузку. Используйте Docker или Kubernetes для автоматизации масштабирования воркеров.

Настройте task_acks_late на True, чтобы задачи подтверждались только после успешного выполнения. Это предотвращает потерю данных при сбоях воркеров.

Используйте task_reject_on_worker_lost для повторной постановки задач в очередь, если воркер завершил работу неожиданно. Это особенно полезно в облачных средах, где серверы могут быть остановлены автоматически.

Для длительных задач настройте task_time_limit и task_soft_time_limit. Установите разумные значения, чтобы задачи не блокировали воркеры слишком долго.

Включите мониторинг с помощью Flower. Это позволит отслеживать состояние задач, воркеров и очередей в реальном времени. Настройте уведомления о сбоях через Slack или Email.

Используйте параметр broker_pool_limit для управления количеством соединений с брокером. Увеличьте его значение, если наблюдаете задержки при обработке задач.

Пример конфигурации для масштабируемого приложения:

Периодически тестируйте конфигурацию под нагрузкой. Используйте инструменты, такие как Locust, чтобы убедиться, что система справляется с пиковыми значениями.