Задачи¶

Связанные задачи¶

Привязка задачи означает, что первым аргументом задачи всегда будет экземпляр задачи (self), как и в привязанных методах Python:

logger = get_task_logger(__name__)

@app.task(bind=True)
def add(self, x, y):
    logger.info(self.request.id)

Связанные задачи нужны для повторных попыток (с помощью app.Task.retry()), для доступа к информации о текущем запросе задачи и для любой дополнительной функциональности, которую вы добавляете в базовые классы пользовательских задач.

Наследование задач¶

Аргумент base декоратора задачи задает базовый класс задачи:

import celery

class MyTask(celery.Task):

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

@app.task(base=MyTask)
def add(x, y):
    raise KeyError()

Автоматическое именование и относительный импорт¶

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

Например, если клиент импортирует модуль "myapp.tasks" как ".tasks", а рабочий импортирует модуль как "myapp.tasks", сгенерированные имена не совпадут, и рабочий выдаст ошибку NotRegistered.

Это также происходит при использовании Django и использовании именования в стиле project.myapp в INSTALLED_APPS:

INSTALLED_APPS = ['project.myapp']

Если вы установите приложение под именем project.myapp, то модуль задач будет импортирован как project.myapp.tasks, поэтому вы должны убедиться, что всегда импортируете задачи, используя одно и то же имя:

>>> from project.myapp.tasks import mytask   # << GOOD

>>> from myapp.tasks import mytask    # << BAD!!!

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

>>> from project.myapp.tasks import mytask
>>> mytask.name
'project.myapp.tasks.mytask'

>>> from myapp.tasks import mytask
>>> mytask.name
'myapp.tasks.mytask'

По этой причине вы должны быть последовательны в том, как вы импортируете модули, и это также является лучшей практикой Python.

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

from module import foo   # BAD!

from proj.module import foo  # GOOD!

Относительные импорты нового стиля в порядке и могут быть использованы:

from .module import foo  # GOOD!

Если вы хотите использовать Celery в проекте, уже активно использующем эти паттерны, и у вас нет времени на рефакторинг существующего кода, вы можете рассмотреть возможность явного указания имен вместо того, чтобы полагаться на автоматическое именование:

@app.task(name='proj.tasks.add')
def add(x, y):
    return x + y

Изменение поведения автоматического именования¶

Есть некоторые случаи, когда автоматическое именование по умолчанию не подходит. Рассмотрим множество задач в разных модулях:

project/
       /__init__.py
       /celery.py
       /moduleA/
               /__init__.py
               /tasks.py
       /moduleB/
               /__init__.py
               /tasks.py

При использовании автоматического именования по умолчанию каждая задача будет иметь сгенерированное имя типа moduleA.tasks.taskA, moduleA.tasks.taskB, moduleB.tasks.test и так далее. Возможно, вы захотите избавиться от присутствия tasks в именах всех задач. Как указывалось выше, вы можете явно задать имена для всех задач, или изменить поведение автоматического именования, переопределив app.gen_task_name(). Продолжая пример, celery.py может содержать:

from celery import Celery

class MyCelery(Celery):

    def gen_task_name(self, name, module):
        if module.endswith('.tasks'):
            module = module[:-6]
        return super().gen_task_name(name, module)

app = MyCelery('main')

Таким образом, каждая задача будет иметь имя типа moduleA.taskA, moduleA.taskB и moduleB.test.

Пример¶

Примером задания для доступа к информации в контексте является:

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(
            self.request))

Аргумент bind означает, что функция будет «связанным методом», так что вы сможете получить доступ к атрибутам и методам экземпляра типа задачи.

Проверка аргументов¶

Celery проверит переданные аргументы при вызове задачи, как это делает Python при вызове обычной функции:

>>> @app.task
... def add(x, y):
...     return x + y

# Calling the task with two arguments works:
>>> add.delay(8, 8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

# Calling the task with only one argument fails:
>>> add.delay(8)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "celery/app/task.py", line 376, in delay
    return self.apply_async(args, kwargs)
  File "celery/app/task.py", line 485, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: add() takes exactly 2 arguments (1 given)

Вы можете отключить проверку аргументов для любой задачи, установив ее атрибут typing в значение False:

>>> @app.task(typing=False)
... def add(x, y):
...     return x + y

# Works locally, but the worker receiving the task will raise an error.
>>> add.delay(8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

Сокрытие конфиденциальной информации в аргументах¶

При использовании task_protocol 2 и выше (по умолчанию с версии 4.0) вы можете переопределить способ представления позиционных аргументов и ключевых слов в журналах и событиях мониторинга с помощью вызывающих аргументов argsrepr и << 2 >>>:

>>> add.apply_async((2, 3), argsrepr='(<secret-x>, <secret-y>)')

>>> charge.s(account, card='1234 5678 1234 5678').set(
...     kwargsrepr=repr({'card': '**** **** **** 5678'})
... ).delay()

Использование пользовательской задержки повторной попытки¶

Когда задача должна быть повторно запущена, она может подождать определенное количество времени, и задержка по умолчанию определяется атрибутом default_retry_delay. По умолчанию она установлена на 3 минуты. Обратите внимание, что единица измерения задержки - секунды (int или float).

Вы также можете указать аргумент countdown в retry(), чтобы отменить это значение по умолчанию.

@app.task(bind=True, default_retry_delay=30 * 60)  # retry in 30 minutes.
def add(self, x, y):
    try:
        something_raising()
    except Exception as exc:
        # overrides the default delay to retry after 1 minute
        raise self.retry(exc=exc, countdown=60)

Автоматические повторные попытки для известных исключений¶

Иногда вы просто хотите повторить выполнение задачи при возникновении определенного исключения.

К счастью, вы можете указать Celery на автоматическое повторение задачи, используя аргумент autoretry_for в декораторе task():

from twitter.exceptions import FailWhaleError

@app.task(autoretry_for=(FailWhaleError,))
def refresh_timeline(user):
    return twitter.refresh_timeline(user)

Если вы хотите указать пользовательские аргументы для внутреннего вызова retry(), передайте декоратору task() аргумент retry_kwargs:

@app.task(autoretry_for=(FailWhaleError,),
          retry_kwargs={'max_retries': 5})
def refresh_timeline(user):
    return twitter.refresh_timeline(user)

Это предоставляется как альтернатива ручной обработке исключений, и в приведенном выше примере будет сделано то же самое, что и при обертывании тела задачи в try … except:

@app.task
def refresh_timeline(user):
    try:
        twitter.refresh_timeline(user)
    except FailWhaleError as exc:
        raise div.retry(exc=exc, max_retries=5)

Если вы хотите автоматически повторять попытку при любой ошибке, просто используйте:

@app.task(autoretry_for=(Exception,))
def x():
    ...

Если ваши задачи зависят от другого сервиса, например, от запроса к API, то целесообразно использовать exponential backoff, чтобы не перегружать сервис своими запросами. К счастью, поддержка Celery автоматического повтора позволяет сделать это легко. Просто укажите аргумент retry_backoff, например, так:

from requests.exceptions import RequestException

@app.task(autoretry_for=(RequestException,), retry_backoff=True)
def x():
    ...

По умолчанию экспоненциальный откат также вводит случайный джиттер, чтобы избежать выполнения всех задач в один и тот же момент. Кроме того, максимальная задержка отката ограничена 10 минутами. Все эти параметры можно настроить с помощью опций, описанных ниже.

Вы также можете установить параметры autoretry_for, max_retries, retry_backoff, retry_backoff_max и retry_jitter в заданиях на основе классов:

class BaseTaskWithRetry(Task):
    autoretry_for = (TypeError,)
    max_retries = 5
    retry_backoff = True
    retry_backoff_max = 700
    retry_jitter = False

Task.autoretry_for¶

Список/кортеж классов исключений. Если во время выполнения задачи возникнет любое из этих исключений, задача будет автоматически повторена. По умолчанию ни одно исключение не будет автоматически перепроведено.

Task.max_retries¶

Число. Максимальное количество повторных попыток перед сдачей. Значение None означает, что задача будет повторять попытки вечно. По умолчанию этот параметр имеет значение 3.

Task.retry_backoff¶

Булево значение или число. Если этот параметр установлен в True, автоповторы будут отложены по правилам exponential backoff. Первая попытка будет иметь задержку в 1 секунду, вторая - в 2 секунды, третья - в 4 секунды, четвертая - в 8 секунд и так далее. (Однако это значение задержки изменяется параметром retry_jitter, если он включен.) Если этот параметр установлен в число, оно используется в качестве коэффициента задержки. Например, если эта опция установлена в 3, первая повторная попытка задержится на 3 секунды, вторая - на 6 секунд, третья - на 12 секунд, четвертая - на 24 секунды и так далее. По умолчанию эта опция установлена в False, и автоповторы не задерживаются.

Task.retry_backoff_max¶

Число. Если включена опция retry_backoff, то она устанавливает максимальную задержку в секундах между авторежимами задач. По умолчанию эта опция установлена на 600, что составляет 10 минут.

Task.retry_jitter¶

Булево значение. Jitter используется для внесения случайности в экспоненциальные задержки обратного хода, чтобы предотвратить одновременное выполнение всех задач в очереди. Если эта опция установлена в True, значение задержки, вычисленное retry_backoff, рассматривается как максимальное, а фактическое значение задержки будет случайным числом между нулем и этим максимумом. По умолчанию эта опция установлена в True.

Общий¶

Task.name¶

Имя, под которым зарегистрировано задание.

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

См. также Имена.

Task.request¶

Если задача выполняется, то здесь будет содержаться информация о текущем запросе. Используется локальное хранилище Thread.

См. Запрос задания.

Task.max_retries¶

Применяется, только если задача вызывает self.retry или если задача украшена аргументом autoretry_for.

Максимальное количество повторных попыток перед тем, как сдаться. Если число повторных попыток превышает это значение, будет вызвано исключение MaxRetriesExceededError.

Примечание

Вы должны вызвать retry() вручную, так как он не будет автоматически повторять попытку при исключении…

По умолчанию установлено значение 3. Значение None отключает лимит повторных попыток, и задача будет повторять попытки до тех пор, пока не добьется успеха.

Task.throws¶

Необязательный кортеж ожидаемых классов ошибок, которые не должны рассматриваться как фактическая ошибка.

Ошибки из этого списка будут сообщены бэкенду результатов как сбой, но рабочий не будет регистрировать событие как ошибку, и трассировка не будет включена.

Пример:

@task(throws=(KeyError, HttpNotFound)):
def get_foo():
    something()

Типы ошибок:

• Ожидаемые ошибки (в Task.throws)

  Зарегистрирован с серьезностью INFO, отслеживание исключено.

• Неожиданные ошибки

  Зафиксировано с серьезностью ERROR, с включенным отслеживанием.

Task.default_retry_delay¶

Время по умолчанию в секундах до повторного выполнения задания. Может быть либо int, либо float. По умолчанию это трехминутная задержка.

Task.rate_limit¶

Установите лимит скорости для данного типа задач (ограничивает количество задач, которые могут быть запущены за определенный промежуток времени). Задачи все равно будут завершаться, когда действует ограничение скорости, но может пройти некоторое время, прежде чем будет разрешен их запуск.

Если это значение None, ограничение скорости не действует. Если это целое число или float, то оно интерпретируется как «задач в секунду».

Ограничения скорости могут быть указаны в секундах, минутах или часах путем добавления к значению «/s», «/m» или «/h». Задания будут равномерно распределены по указанному временному интервалу.

Пример: «100/m» (сто задач в минуту). Это обеспечит минимальную задержку в 600 мс между запуском двух задач на одном и том же экземпляре рабочего.

По умолчанию установлено значение task_default_rate_limit: если не указано, ограничение скорости для задач по умолчанию отключено.

Обратите внимание, что это ограничение скорости для каждого рабочего экземпляра, а не глобальное ограничение скорости. Чтобы обеспечить глобальное ограничение скорости (например, для API с максимальным количеством запросов в секунду), вы должны ограничиться определенной очередью.

Task.time_limit¶

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

Task.soft_time_limit¶

Мягкий лимит времени для этого задания. Если не задано, используется значение по умолчанию.

Task.ignore_result¶

Не храните состояние задачи. Обратите внимание, что это означает, что вы не можете использовать AsyncResult для проверки готовности задачи или получения ее возвращаемого значения.

Task.store_errors_even_if_ignored¶

Если True, ошибки будут сохранены, даже если задача настроена на игнорирование результатов.

Task.serializer¶

Строка, определяющая используемый по умолчанию метод сериализации. По умолчанию используется значение task_serializer. Может быть pickle, json, yaml или любой пользовательский метод сериализации, который был зарегистрирован в kombu.serialization.registry.

Для получения дополнительной информации см. раздел Сериализаторы.

Task.compression¶

Строка, определяющая используемую по умолчанию схему сжатия.

По умолчанию используется значение task_compression. Может быть gzip, или bzip2, или любая пользовательская схема сжатия, зарегистрированная в реестре kombu.compression.

Для получения дополнительной информации см. раздел Компрессия.

Task.backend¶

Бэкенд хранилища результатов, который будет использоваться для этой задачи. Экземпляр одного из классов бэкендов в celery.backends. По умолчанию используется app.backend, определяемый параметром result_backend.

Task.acks_late¶

Если установлено значение True сообщения для этой задачи будут подтверждены после выполнения задачи, а не только перед (поведение по умолчанию).

Примечание: Это означает, что задание может быть выполнено несколько раз, если рабочий сломается в середине выполнения. Убедитесь, что ваши задачи имеют значение idempotent.

Глобальное значение по умолчанию может быть отменено параметром task_acks_late.

Task.track_started¶

Если True, задача будет сообщать о своем статусе как «запущена», когда задача выполняется рабочим. Значение по умолчанию False, так как обычное поведение заключается в том, чтобы не сообщать о таком уровне детализации. Задачи либо ожидают выполнения, либо завершены, либо ожидают повторной попытки. Наличие статуса «начато» может быть полезно, когда есть долго выполняющиеся задачи и есть необходимость сообщить, какая задача выполняется в данный момент.

Имя хоста и идентификатор процесса рабочего, выполняющего задание, будут доступны в метаданных состояния (например, result.info[„pid“])

Глобальное значение по умолчанию может быть отменено параметром task_track_started.

Бэкенды результатов¶

Если вы хотите отслеживать задачи или вам нужны возвращаемые значения, то Celery должен хранить или отправлять состояния куда-то, чтобы их можно было получить позже. Есть несколько встроенных бэкендов результатов на выбор: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc) и Redis - или вы можете определить свой собственный.

Ни один бэкенд не подходит для каждого случая использования. Вам следует ознакомиться с сильными и слабыми сторонами каждого бэкенда и выбрать наиболее подходящий для ваших нужд.

Встроенные государства¶

Нестандартные состояния¶

Вы можете легко определить свои собственные состояния, все, что вам нужно - это уникальное имя. Имя состояния обычно представляет собой строку в верхнем регистре. В качестве примера вы можете посмотреть на abortable tasks, которое определяет пользовательское состояние ABORTED.

Используйте update_state() для обновления состояния задачи:.

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

Здесь я создал состояние «PROGRESS», сообщающее любому приложению, знающему об этом состоянии, что задача в данный момент находится в процессе выполнения, а также о том, на каком этапе процесса она находится, благодаря наличию подсчетов current и total как части мета-данных состояния. Это может быть использовано, например, для создания индикаторов выполнения.

Создание выбираемых исключений¶

Редко известный факт Python заключается в том, что исключения должны соответствовать некоторым простым правилам, чтобы их можно было сериализовать с помощью модуля pickle.

Задачи, вызывающие исключения, которые не являются pickleable, не будут работать должным образом, когда Pickle используется в качестве сериализатора.

Чтобы убедиться в том, что ваши исключения могут быть pickleable, исключение * ДОЛЖНО* предоставлять исходные аргументы, с которыми оно было создано, в своем атрибуте .args. Самый простой способ обеспечить это - сделать так, чтобы исключение вызывало Exception.__init__.

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

# OK:
class HttpError(Exception):
    pass

# BAD:
class HttpError(Exception):

    def __init__(self, status_code):
        self.status_code = status_code

# OK:
class HttpError(Exception):

    def __init__(self, status_code):
        self.status_code = status_code
        Exception.__init__(self, status_code)  # <-- REQUIRED

Поэтому правило таково: Для любого исключения, поддерживающего пользовательские аргументы *args, необходимо использовать Exception.__init__(self, *args).

Специальной поддержки для ключевых аргументов не существует, поэтому если вы хотите сохранить ключевые аргументы при распаковке исключения, вы должны передавать их как обычные args:

class HttpError(Exception):

    def __init__(self, status_code, headers=None, body=None):
        self.status_code = status_code
        self.headers = headers
        self.body = body

        super(HttpError, self).__init__(status_code, headers, body)

Игнорировать¶

Задача может поднять Ignore, чтобы заставить рабочего игнорировать задачу. Это означает, что для задачи не будет записано состояние, но сообщение будет подтверждено (удалено из очереди).

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

Пример хранения отозванных задач в наборе Redis:

from celery.exceptions import Ignore

@app.task(bind=True)
def some_task(self):
    if redis.ismember('tasks.revoked', self.request.id):
        raise Ignore()

Пример, в котором результаты сохраняются вручную:

from celery import states
from celery.exceptions import Ignore

@app.task(bind=True)
def get_tweets(self, user):
    timeline = twitter.get_timeline(user)
    if not self.request.called_directly:
        self.update_state(state=states.SUCCESS, meta=timeline)
    raise Ignore()

Отклонить¶

Задача может поднять Reject, чтобы отклонить сообщение задачи, используя метод AMQPs basic_reject. Это не будет иметь никакого эффекта, если не включено Task.acks_late.

Отклонение сообщения имеет тот же эффект, что и подтверждение, но некоторые брокеры могут реализовать дополнительную функциональность, которую можно использовать. Например, RabbitMQ поддерживает концепцию Dead Letter Exchanges, где очередь может быть настроена на использование обмена мертвыми буквами, на которые повторно доставляются отклоненные сообщения.

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

Пример использования reject, когда задача вызывает состояние out of memory:

import errno
from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def render_scene(self, path):
    file = get_file(path)
    try:
        renderer.render_scene(file)

    # if the file is too big to fit in memory
    # we reject it so that it's redelivered to the dead letter exchange
    # and we can manually inspect the situation.
    except MemoryError as exc:
        raise Reject(exc, requeue=False)
    except OSError as exc:
        if exc.errno == errno.ENOMEM:
            raise Reject(exc, requeue=False)

    # For any other error we retry after 10 seconds.
    except Exception as exc:
        raise self.retry(exc, countdown=10)

Пример повторной регистрации сообщения:

from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def requeues(self):
    if not self.request.delivery_info['redelivered']:
        raise Reject('no reason', requeue=True)
    print('received two times')

Для получения более подробной информации о методе basic_reject обратитесь к документации вашего брокера.

Повторная попытка¶

Исключение Retry поднимается методом Task.retry, чтобы сообщить работнику о повторной попытке выполнения задания.

Instantiation¶

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

Это означает, что конструктор __init__ будет вызываться только один раз для каждого процесса, и что класс задачи семантически ближе к Actor.

Если у вас есть задача,

from celery import Task

class NaiveAuthenticateServer(Task):

    def __init__(self):
        self.users = {'george': 'password'}

    def run(self, username, password):
        try:
            return self.users[username] == password
        except KeyError:
            return False

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

Это также может быть полезно для кэширования ресурсов, например, базовый класс Task, который кэширует соединение с базой данных:

from celery import Task

class DatabaseTask(Task):
    _db = None

    @property
    def db(self):
        if self._db is None:
            self._db = Database.connect()
        return self._db

@app.task(base=DatabaseTask)
def process_rows():
    for row in process_rows.db.table.all():
        process_row(row)

from celery import Celery

app = Celery('tasks', task_cls='your.module.path:DatabaseTask')

Обработчики¶

after_return(self, status, retval, task_id, args, kwargs, einfo)¶

Обработчик, вызываемый после возврата задачи.

Parameters:

• status – Текущее состояние задачи.

• retval – Возвращаемое значение/исключение задачи.

• task_id – Уникальный идентификатор задания.

• args – Исходные аргументы для задачи, которая вернулась.

• kwargs – Исходные аргументы ключевых слов для задачи, которая вернулась.

Именованные аргументы:

einfo – ExceptionInfo экземпляр, содержащий обратную трассировку (если таковая имеется).

Возвращаемое значение этого обработчика игнорируется.

on_failure(self, exc, task_id, args, kwargs, einfo)¶

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

Parameters:

• exc – Исключение, вызванное задачей.

• task_id – Уникальный идентификатор неудачного задания.

• args – Исходные аргументы для задания, которое не удалось выполнить.

• kwargs – Исходные аргументы ключевых слов для задания, которое не удалось выполнить.

Именованные аргументы:

einfo – ExceptionInfo экземпляр, содержащий отслеживание.

Возвращаемое значение этого обработчика игнорируется.

on_retry(self, exc, task_id, args, kwargs, einfo)¶

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

Parameters:

• exc – Исключение, отправленное в retry().

• task_id – Уникальный идентификатор повторного задания.

• args – Исходные аргументы для повторно выполняемого задания.

• kwargs – Исходные аргументы ключевых слов для повторно выполняемого задания.

Именованные аргументы:

einfo – ExceptionInfo экземпляр, содержащий отслеживание.

Возвращаемое значение этого обработчика игнорируется.

on_success(self, retval, task_id, args, kwargs)¶

Выполняется рабочим, если задание выполнено успешно.

Parameters:

• retval – Возвращаемое значение задачи.

• task_id – Уникальный идентификатор выполненного задания.

• args – Исходные аргументы для выполняемого задания.

• kwargs – Исходные аргументы ключевых слов для выполняемого задания.

Возвращаемое значение этого обработчика игнорируется.

Запросы и пользовательские запросы¶

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

Пользовательские классы задач могут переопределить, какой класс запроса использовать, изменив атрибут celery.app.task.Task.Task.Request. Вы можете назначить либо сам класс пользовательского запроса, либо его полное имя.

У запроса есть несколько обязанностей. Пользовательские классы запросов должны покрывать их все - они отвечают за фактический запуск и отслеживание задания. Мы настоятельно рекомендуем наследоваться от celery.worker.request.Request.

При использовании pre-forking worker, методы on_timeout() и on_failure() выполняются в основном процессе worker. Приложение может использовать это средство для обнаружения сбоев, которые не были обнаружены с помощью celery.app.task.Task.on_failure().

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

import logging
from celery import Task
from celery.worker.request import Request

logger = logging.getLogger('my.package')

class MyRequest(Request):
    'A minimal custom request to log failures and hard time limits.'

    def on_timeout(self, soft, timeout):
        super(MyRequest, self).on_timeout(soft, timeout)
        if not soft:
           logger.warning(
               'A hard timeout was enforced for task %s',
               self.task.name
           )

    def on_failure(self, exc_info, send_failed_event=True, return_ok=False):
        super().on_failure(
            exc_info,
            send_failed_event=send_failed_event,
            return_ok=return_ok
        )
        logger.warning(
            'Failure detected for task %s',
            self.task.name
        )

class MyTask(Task):
    Request = MyRequest  # you can use a FQN 'my.package:MyRequest'

@app.task(base=MyTask)
def some_longrunning_task():
    # use your imagination

Игнорируйте результаты, которые вам не нужны¶

Если вам не важны результаты выполнения задачи, обязательно установите опцию ignore_result, так как хранение результатов тратит время и ресурсы.

@app.task(ignore_result=True)
def mytask():
    something()

Результаты могут быть даже отключены глобально с помощью параметра task_ignore_result.

Результаты могут быть включены/выключены на основе каждого выполнения, передавая булевый параметр ignore_result при вызове apply_async или delay.

@app.task
def mytask(x, y):
    return x + y

# No result will be stored
result = mytask.apply_async(1, 2, ignore_result=True)
print result.get() # -> None

# Result will be stored
result = mytask.apply_async(1, 2, ignore_result=False)
print result.get() # -> 3

По умолчанию задачи будут не игнорировать результаты (ignore_result=False), если настроен бэкенд результатов.

Порядок приоритета опций следующий:

1. Глобальный task_ignore_result

2. ignore_result опция

3. Вариант выполнения задания ignore_result

Другие советы по оптимизации¶

Дополнительные советы по оптимизации вы найдете в разделе Optimizing Guide.

Избегайте запуска синхронных подзадач¶

Задача, ожидающая результата другой задачи, действительно неэффективна и может даже вызвать тупик, если пул рабочих исчерпан.

Вместо этого сделайте свою разработку асинхронной, например, используя обратные вызовы.

Плохо:

@app.task
def update_page_info(url):
    page = fetch_page.delay(url).get()
    info = parse_page.delay(url, page).get()
    store_page_info.delay(url, info)

@app.task
def fetch_page(url):
    return myhttplib.get(url)

@app.task
def parse_page(page):
    return myparser.parse_document(page)

@app.task
def store_page_info(url, info):
    return PageInfo.objects.create(url, info)

Хорошо:

def update_page_info(url):
    # fetch_page -> parse_page -> store_page
    chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url)
    chain()

@app.task()
def fetch_page(url):
    return myhttplib.get(url)

@app.task()
def parse_page(page):
    return myparser.parse_document(page)

@app.task(ignore_result=True)
def store_page_info(info, url):
    PageInfo.objects.create(url=url, info=info)

Здесь я вместо этого создал цепочку задач, соединив вместе различные signature()“ы. Вы можете прочитать о цепочках и других мощных конструкциях в Canvas: Проектирование рабочих потоков.

По умолчанию Celery не позволяет синхронно запускать подзадачи внутри задачи, но в редких или экстремальных случаях вам может понадобиться это сделать. ПРЕДУПРЕЖДЕНИЕ: не рекомендуется разрешать подзадачам работать синхронно!

@app.task
def update_page_info(url):
    page = fetch_page.delay(url).get(disable_sync_subtasks=False)
    info = parse_page.delay(url, page).get(disable_sync_subtasks=False)
    store_page_info.delay(url, info)

@app.task
def fetch_page(url):
    return myhttplib.get(url)

@app.task
def parse_page(url, page):
    return myparser.parse_document(page)

@app.task
def store_page_info(url, info):
    return PageInfo.objects.create(url, info)

Зернистость¶

Гранулярность задачи - это объем вычислений, необходимый для каждой подзадачи. В общем случае лучше разбить проблему на множество мелких задач, чем иметь несколько долго выполняющихся задач.

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

Однако выполнение задачи сопряжено с накладными расходами. Необходимо отправить сообщение, данные могут быть не локальными и т.д. Поэтому, если задачи слишком мелкозернистые, добавляемые накладные расходы, вероятно, сводят на нет все преимущества.

Местонахождение данных¶

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

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

Самый простой способ обмена данными между рабочими - использовать распределенную систему кэширования, например memcached.

Государство¶

Поскольку Celery является распределенной системой, вы не можете знать, какой процесс или на какой машине будет выполняться задание. Вы даже не можете знать, будет ли задание выполнено своевременно.

Древние изречения об асинхронности говорят нам, что «утверждение мира - обязанность задачи». Это означает, что картина мира могла измениться с момента запроса задачи, поэтому задача отвечает за то, чтобы убедиться, что мир таков, каким он должен быть; если у вас есть задача, которая переиндексирует поисковую систему, а поисковая система должна переиндексироваться максимум каждые 5 минут, то утверждать это должна задача, а не вызывающая сторона.

Еще одна загвоздка - объекты модели Django. Их не следует передавать в качестве аргументов задачам. Почти всегда лучше повторно получить объект из базы данных во время выполнения задачи, так как использование старых данных может привести к условиям гонки.

Представьте себе следующий сценарий: у вас есть статья и задача, которая автоматически расширяет некоторые сокращения в ней:

class Article(models.Model):
    title = models.CharField()
    body = models.TextField()

@app.task
def expand_abbreviations(article):
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

Сначала автор создает статью и сохраняет ее, затем он нажимает на кнопку, которая инициирует задачу сокращения:

>>> article = Article.objects.get(id=102)
>>> expand_abbreviations.delay(article)

Сейчас очередь очень занята, поэтому задание не будет выполнено еще 2 минуты. За это время другой автор вносит изменения в статью, поэтому, когда задание наконец будет выполнено, тело статьи будет возвращено к старой версии, поскольку в аргументе задания было указано старое тело.

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

@app.task
def expand_abbreviations(article_id):
    article = Article.objects.get(id=article_id)
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

>>> expand_abbreviations.delay(article_id)

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

Транзакции базы данных¶

Давайте посмотрим на другой пример:

from django.db import transaction
from django.http import HttpResponseRedirect

@transaction.atomic
def create_article(request):
    article = Article.objects.create()
    expand_abbreviations.delay(article.pk)
    return HttpResponseRedirect('/articles/')

Это представление Django, создающее объект статьи в базе данных, а затем передающее первичный ключ задаче. Он использует декоратор transaction.atomic, который зафиксирует транзакцию, когда представление вернется, или откатится назад, если представление вызовет исключение.

Если задача начнет выполняться до фиксации транзакции, возникнет состояние гонки; Объект базы данных еще не существует!

Решением является использование обратного вызова on_commit для запуска задачи Celery после успешной фиксации всех транзакций.

from django.db.transaction import on_commit

def create_article(request):
    article = Article.objects.create()
    on_commit(lambda: expand_abbreviations.delay(article.pk))

blog/models.py¶

Модель комментариев выглядит следующим образом:

from django.db import models
from django.utils.translation import ugettext_lazy as _


class Comment(models.Model):
    name = models.CharField(_('name'), max_length=64)
    email_address = models.EmailField(_('email address'))
    homepage = models.URLField(_('home page'),
                               blank=True, verify_exists=False)
    comment = models.TextField(_('comment'))
    pub_date = models.DateTimeField(_('Published date'),
                                    editable=False, auto_add_now=True)
    is_spam = models.BooleanField(_('spam?'),
                                  default=False, editable=False)

    class Meta:
        verbose_name = _('comment')
        verbose_name_plural = _('comments')

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

blog/views.py¶

from django import forms
from django.http import HttpResponseRedirect
from django.template.context import RequestContext
from django.shortcuts import get_object_or_404, render_to_response

from blog import tasks
from blog.models import Comment


class CommentForm(forms.ModelForm):

    class Meta:
        model = Comment


def add_comment(request, slug, template_name='comments/create.html'):
    post = get_object_or_404(Entry, slug=slug)
    remote_addr = request.META.get('REMOTE_ADDR')

    if request.method == 'post':
        form = CommentForm(request.POST, request.FILES)
        if form.is_valid():
            comment = form.save()
            # Check spam asynchronously.
            tasks.spam_filter.delay(comment_id=comment.id,
                                    remote_addr=remote_addr)
            return HttpResponseRedirect(post.get_absolute_url())
    else:
        form = CommentForm()

    context = RequestContext(request, {'form': form})
    return render_to_response(template_name, context_instance=context)

Для фильтрации спама в комментариях я использую Akismet, сервис, используемый для фильтрации спама в комментариях, размещенных на бесплатной блог-платформе Wordpress. Akismet, является бесплатным для личного использования, но для коммерческого использования вам нужно заплатить. Вы должны зарегистрироваться на их сервисе, чтобы получить ключ API.

Для осуществления вызовов API к Akismet я использую библиотеку akismet.py, написанную Michael Foord.

blog/tasks.py¶

from celery import Celery

from akismet import Akismet

from django.core.exceptions import ImproperlyConfigured
from django.contrib.sites.models import Site

from blog.models import Comment


app = Celery(broker='amqp://')


@app.task
def spam_filter(comment_id, remote_addr=None):
    logger = spam_filter.get_logger()
    logger.info('Running spam filter for comment %s', comment_id)

    comment = Comment.objects.get(pk=comment_id)
    current_domain = Site.objects.get_current().domain
    akismet = Akismet(settings.AKISMET_KEY, 'http://{0}'.format(domain))
    if not akismet.verify_key():
        raise ImproperlyConfigured('Invalid AKISMET_KEY')


    is_spam = akismet.comment_check(user_ip=remote_addr,
                        comment_content=comment.comment,
                        comment_author=comment.name,
                        comment_author_email=comment.email_address)
    if is_spam:
        comment.is_spam = True
        comment.save()

    return is_spam