Расширения и этапы запуска

app

Текущий экземпляр приложения.

hostname

Имя узла рабочего процесса (например, worker1@example.com).

blueprint

Это Blueprint рабочего процесса.

hub

Объект цикла событий (Hub). С его помощью можно зарегистрировать обработчики в цикле событий.

Поддерживается только транспортами с включённым асинхронным вводом-выводом (amqp, redis); в этом случае необходимо установить атрибут worker.use_eventloop.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Hub:

class WorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Hub'}

pool

Текущий пул процессов/eventlet/gevent/потоков. См. celery.concurrency.base.BasePool.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Pool:

class WorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Pool'}

timer

Timer, используемый для планирования функций.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Timer:

class WorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Timer'}

statedb

Database <celery.worker.state.Persistent>` для сохранения состояния между перезапусками рабочего процесса.

Определяется только при включённом аргументе statedb.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Statedb:

class WorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Statedb'}

autoscaler

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

Определяется только при включённом аргументе autoscale.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Autoscaler:

class WorkerStep(bootsteps.StartStopStep):
    requires = ('celery.worker.autoscaler:Autoscaler',)

Пример этапа запуска Worker:

from celery import bootsteps

class ExampleWorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Pool'}

    def __init__(self, worker, **kwargs):
        print('Called when the WorkController instance is constructed')
        print('Arguments to WorkController: {0!r}'.format(kwargs))

    def create(self, worker):
        # this method can be used to delegate the action methods
        # to another object that implements ``start`` and ``stop``.
        return self

    def start(self, worker):
        print('Called when the worker is started.')

    def stop(self, worker):
        print('Called when the worker shuts down.')

    def terminate(self, worker):
        print('Called when the worker terminates')

Каждый метод получает текущий экземпляр WorkController в качестве первого аргумента.

В другом примере таймер используется для регулярного пробуждения:

from celery import bootsteps


class DeadlockDetection(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Timer'}

    def __init__(self, worker, deadlock_timeout=3600):
        self.timeout = deadlock_timeout
        self.requests = []
        self.tref = None

    def start(self, worker):
        # run every 30 seconds.
        self.tref = worker.timer.call_repeatedly(
            30.0, self.detect, (worker,), priority=10,
        )

    def stop(self, worker):
        if self.tref:
            self.tref.cancel()
            self.tref = None

    def detect(self, worker):
        # update active requests
        for req in worker.active_requests:
            if req.time_start and time() - req.time_start > self.timeout:
                raise SystemExit()

Рабочий процесс Celery отправляет сообщения в подсистему журналирования Python для различных событий на протяжении всего жизненного цикла задачи. Эти сообщения можно настроить, переопределив форматные строки LOG_<TYPE>, определённые в celery/app/trace.py. Например:

import celery.app.trace

celery.app.trace.LOG_SUCCESS = "This is a custom message"

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

import celery.app.trace

celery.app.trace.LOG_REJECTED = "%(name)r is cursed and I won't run it: %(exc)s"

app

Текущий экземпляр приложения.

controller

Родительский объект WorkController, создавший этот потребитель.

hostname

Имя узла рабочего процесса (например, worker1@example.com).

blueprint

Это Blueprint рабочего процесса.

hub

Объект цикла событий (Hub). С его помощью можно зарегистрировать обработчики в цикле событий.

Поддерживается только транспортами с включённым асинхронным вводом-выводом (amqp, redis); в этом случае необходимо установить атрибут worker.use_eventloop.

Чтобы использовать этот объект, этап запуска рабочего процесса должен требовать этап Hub:

class WorkerStep(bootsteps.StartStopStep):
    requires = {'celery.worker.components:Hub'}

connection

Текущее соединение с брокером (kombu.Connection).

Чтобы использовать этот объект, этап запуска потребителя должен требовать этап «Connection»:

class Step(bootsteps.StartStopStep):
    requires = {'celery.worker.consumer.connection:Connection'}

event_dispatcher

Объект app.events.Dispatcher, который можно использовать для отправки событий.

Чтобы использовать этот объект, этап запуска потребителя должен требовать этап Events.

class Step(bootsteps.StartStopStep):
    requires = {'celery.worker.consumer.events:Events'}

gossip

Широковещательная связь между рабочими процессами (Gossip).

Чтобы использовать этот объект, этап запуска потребителя должен требовать этап Gossip.

class RatelimitStep(bootsteps.StartStopStep):
    """Rate limit tasks based on the number of workers in the
    cluster."""
    requires = {'celery.worker.consumer.gossip:Gossip'}

    def start(self, c):
        self.c = c
        self.c.gossip.on.node_join.add(self.on_cluster_size_change)
        self.c.gossip.on.node_leave.add(self.on_cluster_size_change)
        self.c.gossip.on.node_lost.add(self.on_node_lost)
        self.tasks = [
            self.app.tasks['proj.tasks.add']
            self.app.tasks['proj.tasks.mul']
        ]
        self.last_size = None

    def on_cluster_size_change(self, worker):
        cluster_size = len(list(self.c.gossip.state.alive_workers()))
        if cluster_size != self.last_size:
            for task in self.tasks:
                task.rate_limit = 1.0 / cluster_size
            self.c.reset_rate_limits()
            self.last_size = cluster_size

    def on_node_lost(self, worker):
        # may have processed heartbeat too late, so wake up soon
        # in order to see if the worker recovered.
        self.c.timer.call_after(10.0, self.on_cluster_size_change)

Обработчики

• <set> gossip.on.node_join

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

• <set> gossip.on.node_leave

  Вызывается при каждом выходе нового узла из кластера (завершении работы) и получает экземпляр Worker.

• <set> gossip.on.node_lost

  Вызывается, если для рабочего процесса в кластере не получен сигнал проверки активности (сигнал не получен или не обработан вовремя), и получает экземпляр Worker.

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

pool

Текущий пул процессов/eventlet/gevent/потоков. См. celery.concurrency.base.BasePool.

timer

Timer <celery.utils.timer2.Schedule, используемый для планирования функций.

heart

Отвечает за отправку сигналов проверки активности рабочего процесса (Heart).

Чтобы использовать этот объект, этап запуска потребителя должен требовать этап Heart:

class Step(bootsteps.StartStopStep):
    requires = {'celery.worker.consumer.heart:Heart'}

task_consumer

Объект kombu.Consumer, используемый для получения сообщений задач.

Чтобы использовать этот объект, этап запуска потребителя должен требовать этап Tasks:

class Step(bootsteps.StartStopStep):
    requires = {'celery.worker.consumer.tasks:Tasks'}

strategies

Для каждого зарегистрированного типа задачи в этом словаре есть запись, значение которой используется для выполнения входящего сообщения этого типа задачи (стратегия выполнения задачи). Этот словарь создаётся этапом Tasks при запуске потребителя:

for name, task in app.tasks.items():
    strategies[name] = task.start_strategy(app, consumer)
    task.__trace__ = celery.app.trace.build_tracer(
        name, task, loader, hostname
    )

Чтобы использовать этот словарь, этап запуска потребителя должен требовать этап Tasks:

class Step(bootsteps.StartStopStep):
    requires = {'celery.worker.consumer.tasks:Tasks'}

task_buckets

defaultdict, используемый для поиска ограничения частоты выполнения задачи по её типу. Значениями в этом словаре могут быть None (если ограничение отсутствует) или экземпляр TokenBucket, реализующий consume(tokens) и expected_time(tokens).

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

qos

Объект QoS можно использовать для изменения текущего значения prefetch_count канала задач:

# increment at next cycle
consumer.qos.increment_eventually(1)
# decrement at next cycle
consumer.qos.decrement_eventually(1)
consumer.qos.set(10)

consumer.reset_rate_limits()

Обновляет словарь task_buckets для всех зарегистрированных типов задач.

consumer.bucket_for_task(type, Bucket=TokenBucket)

Создаёт корзину ограничения частоты для задачи, используя её атрибут task.rate_limit.

consumer.add_task_queue(name, exchange=None, exchange_type=None,

routing_key=None, \*\*options):

Добавляет новую очередь для получения сообщений. Изменение сохраняется при перезапуске соединения.

consumer.cancel_task_queue(name)

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

apply_eta_task(request)

Планирует выполнение задачи ETA на основе атрибута request.eta. (Request)

В команду-оболочку celery можно добавлять новые команды с помощью точек входа setuptools.

Точки входа — это специальные метаданные, которые можно добавить в пакеты для регистрации программы setup.py, а после установки считать из системы с помощью модуля importlib.

Для установки дополнительных подкоманд Celery распознаёт точки входа celery.commands; значение точки входа должно указывать на допустимую команду Click.

Так расширение для мониторинга https://pypi.org/project/Flower/ может добавить команду celery flower: для этого в setup.py добавляется точка входа:

setup(
    name='flower',
    entry_points={
        'celery.commands': [
           'flower = flower.command:flower',
        ],
    }
)

Определение команды состоит из двух частей, разделённых знаком равенства: первая часть — имя подкоманды (flower), вторая — полный путь к символу функции, реализующей команду:

flower.command:flower

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

В модуле flower/command.py функцию команды можно определить следующим образом:

import click

@click.command()
@click.option('--port', default=8888, type=int, help='Webserver port')
@click.option('--debug', is_flag=True)
def flower(port, debug):
    print('Running our command')

Hub — асинхронный цикл событий рабочего процесса

поддерживаемые транспорты:

amqp, redis

При использовании транспортов брокера amqp или redis рабочий процесс применяет асинхронный ввод-вывод. В конечном счёте планируется, что все транспорты будут использовать цикл событий, но на это потребуется время, поэтому в остальных транспортах пока используется решение на основе потоков.

hub.add(fd, callback, flags)

hub.add_reader(fd, callback, \*args)

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

Обработчик останется зарегистрированным, пока его явно не удалят с помощью hub.remove(fd) или пока файловый дескриптор не будет автоматически отброшен из-за того, что он больше недействителен.

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

Файловый дескриптор — это любой файловый объект, поддерживающий метод fileno, либо номер файлового дескриптора (int).

hub.add_writer(fd, callback, \*args)

Добавляет обработчик, который будет вызван, когда fd станет доступен для записи. См. также примечания выше для hub.add_reader().

hub.remove(fd)

Удаляет из цикла все обработчики для файлового дескриптора fd.

timer.call_after(secs, callback, args=(), kwargs=(),

priority=0)

timer.call_repeatedly(secs, callback, args=(), kwargs=(),

priority=0)

timer.call_at(eta, callback, args=(), kwargs=(),

priority=0)