Руководство для работников¶

Запуск рабочего¶

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

$ celery -A proj worker -l INFO

Полный список доступных опций командной строки смотрите в worker, или просто сделайте:

$ celery worker --help

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

$ celery -A proj worker --loglevel=INFO --concurrency=10 -n worker1@%h
$ celery -A proj worker --loglevel=INFO --concurrency=10 -n worker2@%h
$ celery -A proj worker --loglevel=INFO --concurrency=10 -n worker3@%h

Аргумент hostname может расширять следующие переменные:

• %h: Имя хоста, включая доменное имя.

• %n: Только имя хоста.

• %d: Только доменное имя.

Если текущее имя хоста - george.example.com, они расширятся до:

Остановка работника¶

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

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

Если рабочий не завершает работу по истечении определенного времени, застряв в бесконечном цикле или подобном, вы можете использовать сигнал KILL для принудительного завершения работы рабочего: но имейте в виду, что текущие выполняющиеся задания будут потеряны (т.е. если у заданий не установлена опция acks_late).

Также, поскольку процессы не могут отменить сигнал KILL, рабочий не сможет получить своих детей; убедитесь, что это можно сделать вручную. Эта команда обычно помогает:

$ pkill -9 -f 'celery worker'

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

$ ps auxww | awk '/celery worker/ {print $2}' | xargs kill -9

Перезапуск рабочего¶

Для перезапуска рабочего необходимо послать сигнал TERM и запустить новый экземпляр. Самый простой способ управления рабочими для разработки - это использование celery multi:

$ celery multi start 1 -A proj -l INFO -c4 --pidfile=/var/run/celery/%n.pid
$ celery multi restart 1 --pidfile=/var/run/celery/%n.pid

Для производственных развертываний следует использовать init-скрипты или систему контроля процессов (см. Демонизация).

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

$ kill -HUP $pid

Сигналы процесса¶

Основной процесс рабочего переопределяет следующие сигналы:

Переменные в путях к файлам¶

Аргументы пути к файлу для --logfile, --pidfile и --statedb могут содержать переменные, которые рабочий будет расширять:

Concurrency¶

По умолчанию для одновременного выполнения задач используется многопроцессорность, но вы также можете использовать Eventlet. Количество рабочих процессов/потоков может быть изменено с помощью аргумента --concurrency и по умолчанию соответствует количеству CPU, доступных на машине.

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

поддержка бассейна:

prefork, eventlet, gevent, thread, блокировка:solo (см. примечание)

поддержка брокеров:

amqp, redis

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

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

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

>>> app.control.broadcast('rate_limit',
...                          arguments={'task_name': 'myapp.mytask',
...                                     'rate_limit': '200/m'})

>>> app.control.broadcast('rate_limit', {
...     'task_name': 'myapp.mytask', 'rate_limit': '200/m'}, reply=True)
[{'worker1.example.com': 'New rate limit set successfully'},
 {'worker2.example.com': 'New rate limit set successfully'},
 {'worker3.example.com': 'New rate limit set successfully'}]

>>> app.control.broadcast('rate_limit', {
...     'task_name': 'myapp.mytask',
...     'rate_limit': '200/m'}, reply=True,
...                             destination=['worker1@example.com'])
[{'worker1.example.com': 'New rate limit set successfully'}]

Команды¶

>>> result.revoke()

>>> AsyncResult(id).revoke()

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed')

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed',
...                    terminate=True)

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed',
...                    terminate=True, signal='SIGKILL')

>>> app.control.revoke([
...    '7993b0aa-1f0b-4780-9af0-c47c0858b3f2',
...    'f565793e-b041-4b2b-9ca4-dca22762a55d',
...    'd9d35e03-2997-42d0-a13e-64a66b88a618',
])

$ celery -A proj worker -l INFO --statedb=/var/run/celery/worker.state

celery multi start 2 -l INFO --statedb=/var/run/celery/%n.state

Ограничения по времени¶

поддержка бассейна:

префорк/гевент

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

Ограничение по времени (–time-limit) - это максимальное количество секунд, в течение которых задача может выполняться до того, как выполняющий ее процесс будет завершен и заменен новым процессом. Вы также можете включить мягкий временной лимит (–soft-time-limit), который вызывает исключение, которое задача может перехватить и устранить до того, как жесткий временной лимит уничтожит ее:

from myapp import app
from celery.exceptions import SoftTimeLimitExceeded

@app.task
def mytask():
    try:
        do_work()
    except SoftTimeLimitExceeded:
        clean_up_in_a_hurry()

Ограничения по времени можно также установить с помощью параметров task_time_limit / << 1 >>>.

>>> app.control.time_limit('tasks.crawl_the_web',
                           soft=60, hard=120, reply=True)
[{'worker1.example.com': {'ok': 'time limits set successfully'}}]

Пределы тарифов¶

>>> app.control.rate_limit('myapp.mytask', '200/m')

>>> app.control.rate_limit('myapp.mytask', '200/m',
...            destination=['celery@worker1.example.com'])

Максимальное количество задач на одно детское учреждение¶

поддержка бассейна:

префорк

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

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

Опция может быть установлена с помощью рабочего аргумента --max-tasks-per-child или с помощью настройки worker_max_tasks_per_child.

Максимальное количество памяти на одну детскую установку¶

поддержка бассейна:

префорк

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

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

Опция может быть установлена с помощью рабочего аргумента --max-memory-per-child или с помощью настройки worker_max_memory_per_child.

Автомасштабирование¶

поддержка бассейна:

префорк, гевент

Компонент autoscaler используется для динамического изменения размера пула в зависимости от нагрузки:

• Автоскалер добавляет дополнительные процессы пула, когда есть работа,

  • и начинает удалять процессы при низкой нагрузке.

Она включается опцией --autoscale, для которой нужны два числа: максимальное и минимальное количество процессов пула:

--autoscale=AUTOSCALE
     Enable autoscaling by providing
     max_concurrency,min_concurrency.  Example:
       --autoscale=10,3 (always keep 3 processes, but grow to
      10 if necessary).

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

Очереди¶

Рабочий экземпляр может потреблять из любого количества очередей. По умолчанию он будет потреблять из всех очередей, определенных в параметре task_queues (который, если не указан, возвращается к очереди по умолчанию с именем celery).

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

$ celery -A proj worker -l INFO -Q foo,bar,baz

Если имя очереди определено в task_queues, то будет использоваться эта конфигурация, но если она не определена в списке очередей, то Celery автоматически сгенерирует для вас новую очередь (в зависимости от опции task_create_missing_queues).

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

$ celery -A proj control add_consumer foo
-> worker1.local: OK
    started consuming from u'foo'

$ celery -A proj control add_consumer foo -d celery@worker1.local

>>> app.control.add_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

>>> app.control.add_consumer('foo', reply=True,
...                          destination=['worker1@example.com'])
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

>>> app.control.add_consumer(
...     queue='baz',
...     exchange='ex',
...     exchange_type='topic',
...     routing_key='media.*',
...     options={
...         'queue_durable': False,
...         'exchange_durable': False,
...     },
...     reply=True,
...     destination=['w1@example.com', 'w2@example.com'])

$ celery -A proj control cancel_consumer foo

$ celery -A proj control cancel_consumer foo -d celery@worker1.local

>>> app.control.cancel_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"no longer consuming from u'foo'"}}]

$ celery -A proj inspect active_queues
[...]

$ celery -A proj inspect active_queues -d celery@worker1.local
[...]

>>> app.control.inspect().active_queues()
[...]

>>> app.control.inspect(['worker1.local']).active_queues()
[...]

Инспектирование работников¶

app.control.inspect позволяет вам проверять работающих рабочих. Он использует команды удаленного управления под капотом.

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

>>> # Inspect all nodes.
>>> i = app.control.inspect()

>>> # Specify multiple nodes to inspect.
>>> i = app.control.inspect(['worker1.example.com',
                            'worker2.example.com'])

>>> # Specify a single node to inspect.
>>> i = app.control.inspect('worker1.example.com')

>>> i.registered()
[{'worker1.example.com': ['tasks.add',
                          'tasks.sleeptask']}]

>>> i.active()
[{'worker1.example.com':
    [{'name': 'tasks.sleeptask',
      'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
      'args': '(8,)',
      'kwargs': '{}'}]}]

>>> i.scheduled()
[{'worker1.example.com':
    [{'eta': '2010-06-07 09:07:52', 'priority': 0,
      'request': {
        'name': 'tasks.sleeptask',
        'id': '1a7980ea-8b19-413e-91d2-0b74f3844c4d',
        'args': '[1]',
        'kwargs': '{}'}},
     {'eta': '2010-06-07 09:07:53', 'priority': 0,
      'request': {
        'name': 'tasks.sleeptask',
        'id': '49661b9a-aa22-4120-94b7-9ee8031d219d',
        'args': '[2]',
        'kwargs': '{}'}}]}]

>>> i.reserved()
[{'worker1.example.com':
    [{'name': 'tasks.sleeptask',
      'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
      'args': '(8,)',
      'kwargs': '{}'}]}]

$ celery -A proj inspect stats

Дополнительные команды¶

>>> app.control.broadcast('shutdown') # shutdown all workers
>>> app.control.broadcast('shutdown', destination='worker1@example.com')

>>> app.control.ping(timeout=0.5)
[{'worker1.example.com': 'pong'},
 {'worker2.example.com': 'pong'},
 {'worker3.example.com': 'pong'}]

>>> ping(['worker2.example.com', 'worker3.example.com'])
[{'worker2.example.com': 'pong'},
 {'worker3.example.com': 'pong'}]

>>> app.control.enable_events()
>>> app.control.disable_events()

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

Существует два типа команд пульта дистанционного управления:

• Команда инспекции

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

• Команда управления

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

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

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

from celery.worker.control import control_command

@control_command(
    args=[('n', int)],
    signature='[N=1]',  # <- used for help on the command-line.
)
def increase_prefetch_count(state, n=1):
    state.consumer.qos.increment_eventually(n)
    return {'ok': 'prefetch count incremented'}

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

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

$ celery -A proj control increase_prefetch_count 3

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

from celery.worker.control import inspect_command

@inspect_command()
def current_prefetch_count(state):
    return {'prefetch_count': state.consumer.qos.value}

После перезапуска рабочего вы можете запросить это значение с помощью программы celery inspect:

$ celery -A proj inspect current_prefetch_count