Расширения и бутстепы¶

Атрибуты¶

app¶

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

hostname¶

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

blueprint¶

Это рабочий Blueprint.

hub¶

Объект цикла событий (Hub). Вы можете использовать его для регистрации обратных вызовов в цикле событий.

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

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

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

pool¶

Текущий пул процессов/событий/гентов/потоков. См. celery.concurrency.base.BasePool.

Ваш рабочий gevent должен требовать использования Pool bootstep:

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',)

autoreloader¶

Autoreloader используется для автоматической перезагрузки кода использования при изменении файловой системы.

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

class WorkerStep(bootsteps.StartStopStep):
    requires = ('celery.worker.autoreloader:Autoreloader',)

Пример рабочего бутстепа¶

Примером Worker bootstep может быть:

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"

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

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.

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

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¶

Текущий пул процессов/событий/гентов/потоков. См. 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) и << 3 >>>.

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

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)

Добавление новых опций командной строки¶

from celery import Celery
from click import Option

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

app.user_options['worker'].add(Option(('--enable-my-option',),
                                      is_flag=True,
                                      help='Enable custom option.'))

from celery import bootsteps

class MyBootstep(bootsteps.Step):

    def __init__(self, parent, enable_my_option=False, **options):
        super().__init__(parent, **options)
        if enable_my_option:
            party()

app.steps['worker'].add(MyBootstep)

from celery import Celery
from celery import signals
from click import Option

app = Celery()

app.user_options['preload'].add(Option(('-Z', '--template'),
                                       default='default',
                                       help='Configuration template to use.'))

@signals.user_preload_options.connect
def on_preload_parsed(options, **kwargs):
    use_template(options['template'])

Добавление новых подкоманд celery¶

Новые команды могут быть добавлены к команде зонтика celery с помощью setuptools entry-points.

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

Celery распознает точки входа celery.commands для установки дополнительных подкоманд, где значение точки входа должно указывать на действительную команду click.

Вот как расширение мониторинга 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)