Следующие шаги¶

Использование Celery в вашем приложении¶

proj/__init__.py
    /celery.py
    /tasks.py

$ celery -A proj worker -l INFO

--------------- celery@halcyon.local v4.0 (latentcall)
--- ***** -----
-- ******* ---- [Configuration]
- *** --- * --- . broker:      amqp://guest@localhost:5672//
- ** ---------- . app:         __main__:0x1012d8590
- ** ---------- . concurrency: 8 (processes)
- ** ---------- . events:      OFF (enable -E to monitor this worker)
- ** ----------
- *** --- * --- [Queues]
-- ******* ---- . celery:      exchange:celery(direct) binding:celery
--- ***** -----

[2012-06-08 16:23:51,078: WARNING/MainProcess] celery@halcyon.local has started.

$ celery worker --help

$ celery multi start w1 -A proj -l INFO
celery multi v4.0.0 (latentcall)
> Starting nodes...
    > w1.halcyon.local: OK

$ celery  multi restart w1 -A proj -l INFO
celery multi v4.0.0 (latentcall)
> Stopping nodes...
    > w1.halcyon.local: TERM -> 64024
> Waiting for 1 node.....
    > w1.halcyon.local: OK
> Restarting node w1.halcyon.local: OK
celery multi v4.0.0 (latentcall)
> Stopping nodes...
    > w1.halcyon.local: TERM -> 64052

$ celery multi stop w1 -A proj -l INFO

$ celery multi stopwait w1 -A proj -l INFO

$ mkdir -p /var/run/celery
$ mkdir -p /var/log/celery
$ celery multi start w1 -A proj -l INFO --pidfile=/var/run/celery/%n.pid \
                                        --logfile=/var/log/celery/%n%I.log

$ celery multi start 10 -A proj -l INFO -Q:1-3 images,video -Q:4,5 data \
    -Q default -L:4,5 debug

Вызов задач¶

Вы можете вызвать задачу с помощью метода delay():

>>> from proj.tasks import add

>>> add.delay(2, 2)

Этот метод на самом деле является сокращением звездообразного аргумента для другого метода под названием apply_async():

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

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

>>> add.apply_async((2, 2), queue='lopri', countdown=10)

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

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

>>> add(2, 2)
4

Эти три метода - delay(), apply_async() и применение (__call__), составляют API вызова Celery, который также используется для подписей.

Более подробный обзор API вызова можно найти в разделе Calling User Guide.

Каждому вызову задачи присваивается уникальный идентификатор (UUID) - это и есть идентификатор задачи.

Методы delay и apply_async возвращают экземпляр AsyncResult, который можно использовать для отслеживания состояния выполнения задач. Но для этого необходимо включить result backend, чтобы состояние можно было где-то хранить.

Результаты отключены по умолчанию, потому что не существует бэкенда результатов, который подходит для всех приложений; чтобы выбрать его, необходимо рассмотреть недостатки каждого отдельного бэкенда. Для многих задач сохранение возвращаемого значения даже не очень полезно, так что это разумное значение по умолчанию. Также обратите внимание, что бэкенды результатов не используются для мониторинга задач и рабочих: для этого Celery использует специальные сообщения о событиях (см. Руководство по мониторингу и управлению).

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

>>> res = add.delay(2, 2)
>>> res.get(timeout=1)
4

Вы можете найти id задачи, посмотрев на атрибут id:

>>> res.id
d6b3aea2-fb9b-4ebc-8da4-848818db9114

Вы также можете просмотреть исключение и traceback, если задача вызвала исключение, на самом деле result.get() будет распространять любые ошибки по умолчанию:

>>> res = add.delay(2, '2')
>>> res.get(timeout=1)

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "celery/result.py", line 221, in get
    return self.backend.wait_for_pending(
  File "celery/backends/asynchronous.py", line 195, in wait_for_pending
    return result.maybe_throw(callback=callback, propagate=propagate)
  File "celery/result.py", line 333, in maybe_throw
    self.throw(value, self._to_remote_traceback(tb))
  File "celery/result.py", line 326, in throw
    self.on_ready.throw(*args, **kwargs)
  File "vine/promises.py", line 244, in throw
    reraise(type(exc), exc, tb)
  File "vine/five.py", line 195, in reraise
    raise value
TypeError: unsupported operand type(s) for +: 'int' and 'str'

Если вы не хотите, чтобы ошибки распространялись, вы можете отключить это, передав propagate:

>>> res.get(propagate=False)
TypeError("unsupported operand type(s) for +: 'int' and 'str'")

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

>>> res.failed()
True

>>> res.successful()
False

Как же он узнает, провалилась задача или нет? Он может узнать это, посмотрев на состояние задачи:

>>> res.state
'FAILURE'

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

PENDING -> STARTED -> SUCCESS

Запущенное состояние - это особое состояние, которое записывается только в том случае, если включен параметр task_track_started, или если для задачи установлен параметр @task(track_started=True).

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

>>> from proj.celery import app

>>> res = app.AsyncResult('this-id-does-not-exist')
>>> res.state
'PENDING'

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

PENDING -> STARTED -> RETRY -> STARTED -> RETRY -> STARTED -> SUCCESS

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

Вызов задач подробно описан в Calling Guide.

Canvas: Проектирование рабочих потоков¶

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

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

Вы можете создать сигнатуру для задачи add, используя аргументы (2, 2), и обратный отсчет 10 секунд следующим образом:

>>> add.signature((2, 2), countdown=10)
tasks.add(2, 2)

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

>>> add.s(2, 2)
tasks.add(2, 2)

>>> s1 = add.s(2, 2)
>>> res = s1.delay()
>>> res.get()
4

# incomplete partial: add(?, 2)
>>> s2 = add.s(2)

# resolves the partial: add(8, 2)
>>> res = s2.delay(8)
>>> res.get()
10

>>> s3 = add.s(2, 2, debug=True)
>>> s3.delay(debug=False)   # debug is now False.

>>> from celery import group
>>> from proj.tasks import add

>>> group(add.s(i, i) for i in range(10))().get()
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

>>> g = group(add.s(i) for i in range(10))
>>> g(10).get()
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]

>>> from celery import chain
>>> from proj.tasks import add, mul

# (4 + 4) * 8
>>> chain(add.s(4, 4) | mul.s(8))().get()
64

>>> # (? + 4) * 8
>>> g = chain(add.s(4) | mul.s(8))
>>> g(4).get()
64

>>> (add.s(4, 4) | mul.s(8))().get()
64

>>> from celery import chord
>>> from proj.tasks import add, xsum

>>> chord((add.s(i, i) for i in range(10)), xsum.s())().get()
90

>>> (group(add.s(i, i) for i in range(10)) | xsum.s())().get()
90

>>> upload_document.s(file) | group(apply_filter.s() for filter in filters)

Маршрутизация¶

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

Настройка task_routes позволяет маршрутизировать задачи по имени и держать все в одном месте:

app.conf.update(
    task_routes = {
        'proj.tasks.add': {'queue': 'hipri'},
    },
)

Вы также можете указать очередь во время выполнения с помощью аргумента queue к apply_async:

>>> from proj.tasks import add
>>> add.apply_async((2, 2), queue='hipri')

Затем вы можете заставить рабочего потреблять из этой очереди, указав опцию celery worker -Q:

$ celery -A proj worker -Q hipri

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

$ celery -A proj worker -Q hipri,celery

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

Чтобы узнать больше о маршрутизации, включая использование всех возможностей маршрутизации AMQP, смотрите Routing Guide.

Дистанционное управление¶

Если вы используете RabbitMQ (AMQP), Redis или Qpid в качестве брокера, то вы можете контролировать и проверять рабочего во время выполнения.

Например, вы можете увидеть, над какими заданиями в настоящее время работает сотрудник:

$ celery -A proj inspect active

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

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

$ celery -A proj inspect active --destination=celery@example.com

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

Команда celery inspect содержит команды, которые ничего не меняют в рабочем; она только возвращает информацию и статистику о том, что происходит внутри рабочего. Список команд инспекции, которые вы можете выполнить:

$ celery -A proj inspect --help

Затем есть команда celery control, которая содержит команды, которые фактически изменяют вещи в рабочем во время выполнения:

$ celery -A proj control --help

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

$ celery -A proj control enable_events

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

$ celery -A proj events --dump

или вы можете запустить интерфейс curses:

$ celery -A proj events

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

$ celery -A proj control disable_events

Команда celery status также использует команды удаленного управления и показывает список online рабочих в кластере:

$ celery -A proj status

Подробнее о команде celery и мониторинге вы можете прочитать в Monitoring Guide.

Часовой пояс¶

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

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

app.conf.timezone = 'Europe/London'

Оптимизация¶

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

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

Если вы используете RabbitMQ, вы можете установить модуль librabbitmq, AMQP-клиент, реализованный на языке C:

$ pip install librabbitmq

Что делать дальше?¶

Теперь, когда вы прочитали этот документ, вам следует перейти к User Guide.

Есть также API reference, если вам так хочется.