Что нового в Celery 3.0 (Chiastic Slide)¶

Переименование вещательных бирж¶

Команда обмена рабочими пультами была переименована (новое имя pidbox), это связано с тем, что флаг auto_delete на обмене был удален, и это делает его несовместимым с более ранними версиями.

При желании вы можете вручную удалить старые обмены, используя команду celery amqp (ранее называлась camqadm):

$ celery amqp exchange.delete celeryd.pidbox
$ celery amqp exchange.delete reply.celeryd.pidbox

Event-loop¶

Рабочий теперь работает без потоков при использовании с RabbitMQ (AMQP) или Redis в качестве брокера, в результате чего:

• Значительно лучшая общая производительность.

• Исправляет несколько граничных условий гонки.

• Субмиллисекундная точность таймера.

• Более быстрое время выключения.

Поддерживаются следующие транспорты: py-amqp librabbitmq, redis и amqplib. Надеемся, что в будущем это можно будет расширить и включить дополнительные брокерские транспорты.

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

Новая команда celery зонтик¶

Все программы командной строки Celery теперь доступны из одной команды celery umbrella.

Список подкоманд и опций можно посмотреть, выполнив команду:

$ celery help

Команды включают:

• celery worker (ранее celeryd).

• celery beat (ранее celerybeat).

• celery amqp (ранее camqadm).

Старые программы все еще доступны (celeryd, celerybeat и т.д.), но вам не рекомендуют их использовать.

Теперь зависит от billiard¶

Billiard - это форк мультипроцессинга, содержащий патч no-execv от sbt (http://bugs.python.org/issue8713), а также содержит улучшения пула, ранее находившиеся в Celery.

Это развилка была необходима, так как для работы патча no-execv требовались изменения в коде расширения C.

• Выпуск #625

• Выпуск #627

• Выпуск #640

• django-celery #122 <https://github.com/celery/django-celery/issues/122

• django-celery #124 <https://github.com/celery/django-celery/issues/122

celery.app.task больше нет пакета¶

Модуль celery.app.task теперь является модулем, а не пакетом.

Сценарий установки setup.py попытается удалить старый пакет, но если по какой-то причине это не сработает, вам придется удалить его вручную. Эта команда поможет:

$ rm -r $(dirname $(python -c 'import celery;print(celery.__file__)'))/app/task/

Если возникает ошибка типа ImportError: cannot import name _unpickle_task, нужно просто удалить старый пакет, и все будет в порядке.

Последняя версия с поддержкой Python 2.5¶

Серия 3.0 будет последней версией, поддерживающей Python 2.5, а начиная с версии 3.1 потребуется Python 2.6 и выше.

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

На данный момент Python 2.6 должен быть широко доступен, и мы призываем вас перейти на него, но если это невозможно, у вас все еще есть возможность продолжать использовать Celery 3.0, а важные исправления, внесенные в Celery 3.1, будут перенесены в Celery 3.0 по запросу.

Теперь используется часовой пояс UTC¶

Это означает, что ETA/countdown в сообщениях не совместимы с версиями Celery до 2.5.

Вы можете отключить UTC и вернуться к старому местному времени, установив параметр CELERY_ENABLE_UTC.

Redis: Улучшения эмуляции Ack¶

Снижение вероятности потери данных.

Акки теперь реализуются путем хранения копии сообщения при его потреблении. Копия не удаляется до тех пор, пока потребитель не подтвердит или не отклонит его.

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

• Тайм-аут видимости

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

  По умолчанию тайм-аут установлен на один час, но его можно изменить, настроив транспортный параметр:

  BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 18000}  # 5 hours
  

Примечание

Сообщения, которые не были подтверждены, будут доставлены повторно, если превышен таймаут видимости. Для пользователей Celery это означает, что задачи ETA/countdown, запланированные на время, превышающее таймаут видимости, будут выполнены дважды (или более). Если вы планируете использовать длительные ETA/countdowns, вам следует соответствующим образом настроить таймаут видимости.

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

Цепочка задач¶

Задачи теперь могут иметь обратные и ошибочные вызовы, а зависимости регистрируются

• Формат сообщений о задачах был обновлен двумя новыми ключами расширения

  Оба ключа могут быть пустыми/неопределенными или списком подзадач.

  • callbacks

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

  • errbacks

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

  • Ключевые аргументы link и link_error были добавлены в apply_async.

    Они добавляют в задачу обратные вызовы и возвраты, подробнее о них можно прочитать в Связывание (обратные вызовы/возвраты).

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

    • задача.запрос.дети

      Содержит экземпляры результатов подзадач, которые применила текущая выполняемая задача.

    • AsyncResult.children

      Возвращает зависимости задач в виде списка AsyncResult/ResultSet экземпляров.

    • AsyncResult.iterdeps

      Рекурсивно перебирает зависимости задач, выдавая кортежи (parent, node).

      Вызывает IncompleteStream, если какая-либо из зависимостей еще не вернулась.

    • AsyncResult.graph

      DependencyGraph> зависимостей задач. С его помощью можно также преобразовать в формат dot:

      with open('graph.dot') as fh:
          result.graph.to_dot(fh)
      

      затем создайте изображение графика:

      $ dot -Tpng graph.dot -o graph.png
      

• Также включена новая специальная подзадача chain:

  >>> from celery import chain
  
  # (2 + 2) * 8 / 2
  >>> res = chain(add.subtask((2, 2)),
                  mul.subtask((8,)),
                  div.subtask((2,))).apply_async()
  >>> res.get() == 16
  
  >>> res.parent.get() == 32
  
  >>> res.parent.parent.get() == 4
  

• Добавляет AsyncResult.get_leaf()

  Ожидает и возвращает результат подзадачи leaf. Это последний узел, найденный при обходе графа, но это означает, что граф может быть только одномерным (по сути, список).

• Добавляет subtask.link(subtask) + subtask.link_error(subtask)

  Сокращение до s.options.setdefault('link', []).append(subtask)

• Добавляет subtask.flatten_links()

  Возвращает уплощенный список всех зависимостей (рекурсивно)

Redis: Приоритетная поддержка¶

Поле сообщения priority теперь соблюдается транспортом Redis путем создания нескольких списков для каждой именованной очереди. Очереди затем потребляются в порядке приоритета.

Поле приоритета представляет собой число в диапазоне от 0 до 9, где 0 - это приоритет по умолчанию и наивысший приоритет.

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

>>> BROKER_TRANSPORT_OPTIONS = {
...     'priority_steps': [0, 2, 4, 6, 8, 9],
... }

Приоритеты, реализованные таким образом, не так надежны, как приоритеты на стороне сервера, поэтому эта функция получила название «квази-приоритеты»; Использование маршрутизации все еще является предлагаемым способом обеспечения качества обслуживания, поскольку приоритеты, реализованные на стороне клиента, не работают в нескольких случаях, например, если рабочий занят долго выполняющимися задачами, предварительно отправил много сообщений, или очереди перегружены.

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

При участии Германа М. Браво.

Redis: Теперь очереди цикличны, чтобы потребление было справедливым¶

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

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

группа/аккорд/цепочка теперь являются подзадачами¶

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

• В задачи добавлен новый ярлык:

  >>> task.s(arg1, arg2, kw=1)
  

  как ярлык:

  >>> task.subtask((arg1, arg2), {'kw': 1})
  

• Задачи можно объединять в цепочки с помощью оператора |:

  >>> (add.s(2, 2), pow.s(2)).apply_async()
  

• Подзадачи могут быть «оценены» с помощью оператора ~:

  >>> ~add.s(2, 2)
  4
  
  >>> ~(add.s(2, 2) | pow.s(2))
  

  это то же самое, что:

  >>> chain(add.s(2, 2), pow.s(2)).apply_async().get()
  

• В словарь подзадач добавлен новый ключ subtask_type.

  Это может быть строка "chord", "group", "chain", "chunks", "xmap" или "xstarmap".

• maybe_subtask теперь использует subtask_type для реконструкции объекта, чтобы использовать при использовании непиксельных сериализаторов.

• Логика этих операций была перенесена в специальные задачи celery.chord, celery.chain и celery.group.

• подзадача больше не наследуется от AttributeDict.

  Теперь это чистый подкласс dict со свойствами для доступа атрибутов к соответствующим ключам.

• Теперь repr’s выводит последовательность, как хотелось бы императивно:

  >>> from celery import chord
  
  >>> (chord([add.s(i, i) for i in xrange(10)], xsum.s())
        | pow.s(2))
  tasks.xsum([tasks.add(0, 0),
              tasks.add(1, 1),
              tasks.add(2, 2),
              tasks.add(3, 3),
              tasks.add(4, 4),
              tasks.add(5, 5),
              tasks.add(6, 6),
              tasks.add(7, 7),
              tasks.add(8, 8),
              tasks.add(9, 9)]) | tasks.pow(2)
  

Новые команды пульта дистанционного управления¶

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

• add_consumer/cancel_consumer

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

  Эти команды доступны программно как app.control.add_consumer() / app.control.cancel_consumer():

  >>> celery.control.add_consumer(queue_name,
  ...     destination=['w1.example.com'])
  >>> celery.control.cancel_consumer(queue_name,
  ...     destination=['w1.example.com'])
  

  или с помощью команды celery control:

  $ celery control -d w1.example.com add_consumer queue
  $ celery control -d w1.example.com cancel_consumer queue
  

  Примечание

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

• autoscale

  Сообщает рабочим, у которых включено --autoscale, об изменении настроек автомасштабирования макс/мин параллелизма.

  Эта команда доступна программно как app.control.autoscale():

  >>> celery.control.autoscale(max=10, min=5,
  ...     destination=['w1.example.com'])
  

  или с помощью команды celery control:

  $ celery control -d w1.example.com autoscale 10 5
  

• pool_grow/pool_shrink

  Приказывает рабочим добавить или удалить процессы пула.

  Эти команды доступны программно как app.control.pool_grow() / app.control.pool_shrink():

  >>> celery.control.pool_grow(2, destination=['w1.example.com'])
  >>> celery.contorl.pool_shrink(2, destination=['w1.example.com'])
  

  или с помощью команды celery control:

  $ celery control -d w1.example.com pool_grow 2
  $ celery control -d w1.example.com pool_shrink 2
  

• celery control теперь поддерживает команды rate_limit и time_limit.

  Подробнее см. в разделе celery control --help.

Crontab теперь поддерживает аргументы «день месяца» и «месяц года».¶

Обновленный список примеров см. по адресу Расписания Crontab.

Неизменяемые подзадачи¶

subtask“ теперь могут быть неизменяемыми, что означает, что аргументы не будут изменены при вызове обратных вызовов:

>>> chain(add.s(2, 2), clear_static_electricity.si())

означает, что он не получит аргумент родительской задачи, а .si() - это сокращение для:

>>> clear_static_electricity.subtask(immutable=True)

Улучшения ведения журнала¶

Поддержка протоколирования теперь лучше соответствует лучшим практикам.

• Классы, используемые рабочим, больше не используют app.get_default_logger, а используют celery.utils.log.get_logger, который просто получает логгер, не устанавливая уровень, и добавляет NullHandler.

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

• Все регистраторы наследуются от общего регистратора под названием «celery».

• Раньше task.get_logger устанавливал новый регистратор для каждой задачи и даже устанавливал уровень журнала. Теперь этого не происходит.

  • Вместо этого все регистраторы задач теперь наследуются от общего регистратора «celery.task», который устанавливается, когда программы вызывают setup_logging_subsystem.

  • Вместо использования LoggerAdapter для дополнения форматера полем task_id и task_name, регистратор базы задач теперь использует специальный форматер, добавляющий эти значения во время выполнения из текущей выполняющейся задачи.

• На самом деле, task.get_logger больше не рекомендуется, лучше добавить в модуль задач логгер на уровне модуля.

  Например, вот так:

  from celery.utils.log import get_task_logger
  
  logger = get_task_logger(__name__)
  
  @celery.task
  def add(x, y):
      logger.debug('Adding %r + %r' % (x, y))
      return x + y
  

  Полученный регистратор будет наследоваться от регистратора "celery.task", так что имя и id текущей задачи будут включены в вывод журнала.

• Перенаправленный вывод из stdout/stderr теперь регистрируется в журнале «celery.redirected».

• Кроме того, несколько warnings.warn были заменены на logger.warn.

• Теперь можно избежать предупреждения „no handlers for logger multiprocessing“.

Реестр задач больше не является глобальным¶

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

Вы можете заставить приложения совместно использовать реестры, указав это:

>>> app1 = Celery()
>>> app2 = Celery(tasks=app1.tasks)

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

@celery.task(shared=False)
def add(x, y):
    return x + y

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

Класс Task больше не привязывается к приложению по умолчанию, он будет впервые привязан (и настроен) при создании конкретного подкласса.

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

from celery.task import Task

class DebugTask(Task):
    abstract = True

    def __call__(self, *args, **kwargs):
        print('CALLING %r' % (self,))
        return self.run(*args, **kwargs)

>>> DebugTask
<unbound DebugTask>

>>> @celery1.task(base=DebugTask)
... def add(x, y):
...     return x + y
>>> add.__class__
<class add of <Celery default:0x101510d10>>

Декораторы ленивых задач¶

Декоратор @task теперь ленив при использовании с пользовательскими приложениями.

То есть, если включено accept_magic_kwargs (ее называют «compat mode»), декоратор задачи выполняется inline, как и раньше, однако для пользовательских приложений декоратор @task теперь возвращает специальный объект PromiseProxy, который оценивается только при доступе.

Все обещания будут оценены при вызове app.finalize() или неявно при первом использовании реестра задач.

Интеллектуальная опция –app¶

Опция --app теперь «автоопределяется».

• Если указанный путь является модулем, он пытается получить атрибут с именем „celery“.

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

Например, если у вас есть проект с именем proj, в котором приложение celery находится в from proj.celery import app, то следующее будет эквивалентно:

$ celery worker --app=proj
$ celery worker --app=proj.celery:
$ celery worker --app=proj.celery:app

Другие новости¶

• Новый CELERYD_WORKER_LOST_WAIT для управления тайм-аутом в секундах перед тем, как billiard.WorkerLostError будет поднят, когда рабочий не может получить сигнал (проблема #595).

  При участии Брендона Кроуфорда.

• Очереди монитора событий Redis теперь удаляются автоматически (проблема #436).

• Методы фабрики экземпляров приложений были преобразованы в кэшированные дескрипторы, создающие новый подкласс при доступе.

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

  class Worker(app.Worker):
      ...
  

• Новый сигнал: task_success.

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

• Теперь экземпляр Celery может быть создан с помощью URL-адреса брокера

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

• Бэкенды результатов теперь можно установить с помощью URL-адреса

  В настоящее время поддерживается только redis. Пример использования:

  CELERY_RESULT_BACKEND = 'redis://localhost/1'
  

• Частота сердцебиения теперь каждые 5 с, а частота отправляется с событием

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

• Модуль celery.actors был удален и вместо него будет частью cl.

• Представляет новую команду celery, которая является точкой входа для всех остальных команд.

  Главная для этой команды может быть запущена вызовом celery.start().

• Аннотации теперь поддерживают декораторы, если ключ начинается с „@“.

  Например:

  def debug_args(fun):
  
      @wraps(fun)
      def _inner(*args, **kwargs):
          print('ARGS: %r' % (args,))
      return _inner
  
  CELERY_ANNOTATIONS = {
      'tasks.add': {'@__call__': debug_args},
  }
  

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

• Отчет об ошибках теперь доступен как команда и команда трансляции

  • Получите его из Python REPL:

    >>> import celery
    >>> print(celery.bugreport())
    

  • Использование программы командной строки celery:

    $ celery report
    

  • Получайте ее от удаленных работников:

    $ celery inspect report
    

• Модуль celery.log перемещен в celery.app.log.

• Модуль celery.task.control перемещен в celery.app.control.

• Новый сигнал: task_revoked

  Отправляется в основной процесс, когда задание отзывается или завершается.

• AsyncResult.task_id переименовано в AsyncResult.id

• TasksetResult.taskset_id переименовано в .id

• xmap(task, sequence) и xstarmap(task, sequence).

  Возвращает список результатов применения функции задачи к каждому элементу последовательности.

  Пример:

  >>> from celery import xstarmap
  
  >>> xstarmap(add, zip(range(10), range(10)).apply_async()
  [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
  

• chunks(task, sequence, chunksize)

• group.skew(start=, stop=, step=)

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

  >>> g = group(add.s(i, i) for i in xrange(10))
  

  Перекос заданий от 0 секунд до 10 секунд:

  >>> g.skew(stop=10)
  

  Первая задача будет выполняться за 0 секунд, вторая - за 1 секунду, третья - за 2 секунды и так далее.

• 99% покрытие тестов

• CELERY_QUEUES теперь может быть списком/кортежем экземпляров Queue.

  Внутри app.amqp.queues теперь является отображением имени/экземпляров Que, вместо преобразования на лету.

• Теперь можно указать соединение для app.control.inspect.

  from kombu import Connection
  
  i = celery.control.inspect(connection=Connection('redis://'))
  i.active_queues()
  

• CELERY_FORCE_EXECV теперь включено по умолчанию.

  Если требуется старое поведение, параметр может быть установлен на False, или новая опция –no-execv на celery worker.

• Устаревший модуль celery.conf был удален.

• Теперь CELERY_TIMEZONE всегда требует установки библиотеки pytz (за исключением случаев, когда часовой пояс установлен на UTC).

• Бэкенд Tokyo Tyrant был удален и больше не поддерживается.

• Теперь используется maybe_declare() для кэширования объявлений очередей.

• Больше нет глобального значения по умолчанию для параметра CELERYBEAT_MAX_LOOP_INTERVAL, вместо этого он устанавливается отдельными планировщиками.

• Worker: теперь усекает очень длинные тела сообщений в отчетах об ошибках.

• Больше не происходит глубокого копирования исключений при попытке сериализации ошибок.

• CELERY_BENCH переменная окружения, теперь также будет выводить статистику использования памяти при завершении работы.

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

• Исключительные аргументы теперь надежно замаринованы

  При участии Мэтта Лонга.

• Worker/Beat больше не регистрирует баннер запуска.

  Раньше это записывалось в журнал с предупреждением о серьезности, а теперь только в stdout.

• Каталог contrib/ в дистрибутиве был переименован в extra/.

• Новый сигнал: task_revoked

• celery.contrib.migrate: Множество улучшений, включая; фильтрацию, миграцию очередей и поддержку подтверждения сообщений на брокере, с которого осуществляется миграция.

  Внесено Джоном Уотсоном.

• Рабочий: Приращения количества префетчей теперь оптимизированы и сгруппированы.

• Рабочий: Больше не вызывает consume в очереди команд дистанционного управления дважды.

  Возможно, это не вызвало никаких проблем, но было лишним.

Интерны¶

• app.broker_connection теперь app.connection

  Оба имени по-прежнему работают.

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

  Этими модулями являются celery.messaging, celery.log, celery.decorators и celery.registry.

• celery.utils рефакторинг на несколько модулей:

  celery.utils.text celery.utils.imports celery.utils.functional

• Теперь вместо kombu.utils.encoding используется celery.utils.encoding.

• Переименован модуль celery.routes -> celery.app.routes.

• Переименован пакет celery.db -> celery.backends.database.

• Переименован модуль celery.abstract -> celery.worker.bootsteps.

• Документы командной строки теперь анализируются из документов модуля.

• Каталог тестовых наборов был реорганизован.

• setup.py теперь читает документы из каталога requirements/.

• Команды Celery больше не заворачивают вывод (проблема #700).

  Внесено Томасом Йоханссоном.

celery.contrib.methods: Декоратор задач для методов¶

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

from celery.contrib.methods import task_method

class Counter(object):

    def __init__(self):
        self.value = 1

    @celery.task(name='Counter.increment', filter=task_method)
    def increment(self, n=1):
        self.value += 1
        return self.value

Дополнительную информацию см. в разделе celery.contrib.methods.