你正在阅读 Celery 3.1 的文档。开发版本文档见： 此处.

Signals

Signals allows decoupled applications to receive notifications when certain actions occur elsewhere in the application.

Celery ships with many signals that you application can hook into to augment behavior of certain actions.

Basics

Several kinds of events trigger signals, you can connect to these signals to perform actions as they trigger.

Example connecting to the after_task_publish signal:

from celery.signals import after_task_publish

@after_task_publish.connect
def task_sent_handler(sender=None, body=None, **kwargs):
    print('after_task_publish for task id {body[id]}'.format(
        body=body,
    ))

Some signals also have a sender which you can filter by. For example the after_task_publish signal uses the task name as a sender, so by providing the sender argument to connect you can connect your handler to be called every time a task with name “proj.tasks.add” is published:

@after_task_publish.connect(sender='proj.tasks.add')
def task_sent_handler(sender=None, body=None, **kwargs):
    print('after_task_publish for task id {body[id]}'.format(
        body=body,
    ))

Signals use the same implementation as django.core.dispatch. As a result other keyword parameters (e.g. signal) are passed to all signal handlers by default.

The best practice for signal handlers is to accept arbitrary keyword arguments (i.e. **kwargs). That way new celery versions can add additional arguments without breaking user code.

Signals

Task Signals

before_task_publish

3.1 新版功能.

Dispatched before a task is published. Note that this is executed in the process sending the task.

Sender is the name of the task being sent.

Provides arguements:

• body

  Task message body.

  This is a mapping containing the task message fields (see Task Messages).

• exchange

  Name of the exchange to send to or a Exchange object.

• routing_key

  Routing key to use when sending the message.

• headers

  Application headers mapping (can be modified).

• properties

  Message properties (can be modified)

• declare

  List of entities (Exchange, Queue or :class:~`kombu.binding` to declare before publishing the message. Can be modified.

• retry_policy

  Mapping of retry options. Can be any argument to kombu.Connection.ensure() and can be modified.

after_task_publish

Dispatched when a task has been sent to the broker. Note that this is executed in the process that sent the task.

Sender is the name of the task being sent.

Provides arguments:

• body

  The task message body, see Task Messages for a reference of possible fields that can be defined.

• exchange

  Name of the exchange or Exchange object used.

• routing_key

  Routing key used.

task_prerun

Dispatched before a task is executed.

Sender is the task class being executed.

Provides arguments:

• task_id

  Id of the task to be executed.

• task

  The task being executed.

• args

  the tasks positional arguments.

• kwargs

  The tasks keyword arguments.

task_postrun

Dispatched after a task has been executed.

Sender is the task class executed.

Provides arguments:

• task_id

  Id of the task to be executed.

• task

  The task being executed.

• args

  The tasks positional arguments.

• kwargs

  The tasks keyword arguments.

• retval

  The return value of the task.

• state

  Name of the resulting state.

task_success

Dispatched when a task succeeds.

Sender is the task class executed.

Provides arguments

• result

  Return value of the task.

task_failure

Dispatched when a task fails.

Sender is the task class executed.

Provides arguments:

• task_id

  Id of the task.

• exception

  Exception instance raised.

• args

  Positional arguments the task was called with.

• kwargs

  Keyword arguments the task was called with.

• traceback

  Stack trace object.

• einfo

  The celery.datastructures.ExceptionInfo instance.

task_revoked

Dispatched when a task is revoked/terminated by the worker.

Sender is the task class revoked/terminated.

Provides arguments:

• request

  This is a Request instance, and not task.request. When using the prefork pool this signal is dispatched in the parent process, so task.request is not available and should not be used. Use this object instead, which should have many of the same fields.

• terminated

  Set to True if the task was terminated.

• signum

  Signal number used to terminate the task. If this is None and terminated is True then TERM should be assumed.

• expired Set to True if the task expired.

App Signals

import_modules

This signal is sent when a program (worker, beat, shell) etc, asks for modules in the CELERY_INCLUDE and CELERY_IMPORTS settings to be imported.

Sender is the app instance.

Worker Signals

celeryd_after_setup

This signal is sent after the worker instance is set up, but before it calls run. This means that any queues from the -Q option is enabled, logging has been set up and so on.

It can be used to e.g. add custom queues that should always be consumed from, disregarding the -Q option. Here’s an example that sets up a direct queue for each worker, these queues can then be used to route a task to any specific worker:

from celery.signals import celeryd_after_setup

@celeryd_after_setup.connect
def setup_direct_queue(sender, instance, **kwargs):
    queue_name = '{0}.dq'.format(sender)  # sender is the nodename of the worker
    instance.app.amqp.queues.select_add(queue_name)

Provides arguments:

• sender Hostname of the worker.

• instance

  This is the celery.apps.worker.Worker instance to be initialized. Note that only the app and hostname (nodename) attributes have been set so far, and the rest of __init__ has not been executed.

• conf

  The configuration of the current app.

celeryd_init

This is the first signal sent when celery worker starts up. The sender is the host name of the worker, so this signal can be used to setup worker specific configuration:

from celery.signals import celeryd_init

@celeryd_init.connect(sender='worker12@example.com')
def configure_worker12(conf=None, **kwargs):
    conf.CELERY_DEFAULT_RATE_LIMIT = '10/m'

or to set up configuration for multiple workers you can omit specifying a sender when you connect:

from celery.signals import celeryd_init

@celeryd_init.connect
def configure_workers(sender=None, conf=None, **kwargs):
    if sender in ('worker1@example.com', 'worker2@example.com'):
        conf.CELERY_DEFAULT_RATE_LIMIT = '10/m'
    if sender == 'worker3@example.com':
        conf.CELERYD_PREFETCH_MULTIPLIER = 0

Provides arguments:

• sender Nodename of the worker.

• instance

  This is the celery.apps.worker.Worker instance to be initialized. Note that only the app and hostname (nodename) attributes have been set so far, and the rest of __init__ has not been executed.

• conf

  The configuration of the current app.

• options

  Options passed to the worker from command-line arguments (including defaults).

worker_init

Dispatched before the worker is started.

worker_ready

Dispatched when the worker is ready to accept work.

worker_process_init

Dispatched in all pool child processes when they start.

Note that handlers attached to this signal must not be blocking for more than 4 seconds, or the process will be killed assuming it failed to start.

worker_process_shutdown

Dispatched in all pool child processes just before they exit.

Note: There is no guarantee that this signal will be dispatched, similarly to finally blocks it’s impossible to guarantee that handlers will be called at shutdown, and if called it may be interrupted during.

Provides arguments:

• pid

  The pid of the child process that is about to shutdown.

• exitcode

  The exitcode that will be used when the child process exits.

worker_shutdown

Dispatched when the worker is about to shut down.

Beat Signals

beat_init

Dispatched when celery beat starts (either standalone or embedded). Sender is the celery.beat.Service instance.

beat_embedded_init

Dispatched in addition to the beat_init signal when celery beat is started as an embedded process. Sender is the celery.beat.Service instance.

Eventlet Signals

eventlet_pool_started

Sent when the eventlet pool has been started.

Sender is the celery.concurrency.eventlet.TaskPool instance.

eventlet_pool_preshutdown

Sent when the worker shutdown, just before the eventlet pool is requested to wait for remaining workers.

Sender is the celery.concurrency.eventlet.TaskPool instance.

eventlet_pool_postshutdown

Sent when the pool has been joined and the worker is ready to shutdown.

Sender is the celery.concurrency.eventlet.TaskPool instance.

eventlet_pool_apply

Sent whenever a task is applied to the pool.

Sender is the celery.concurrency.eventlet.TaskPool instance.

Provides arguments:

• target

  The target function.

• args

  Positional arguments.

• kwargs

  Keyword arguments.

Logging Signals

setup_logging

Celery won’t configure the loggers if this signal is connected, so you can use this to completely override the logging configuration with your own.

If you would like to augment the logging configuration setup by Celery then you can use the after_setup_logger and after_setup_task_logger signals.

Provides arguments:

• loglevel

  The level of the logging object.

• logfile

  The name of the logfile.

• format

  The log format string.

• colorize

  Specify if log messages are colored or not.

after_setup_logger

Sent after the setup of every global logger (not task loggers). Used to augment logging configuration.

Provides arguments:

• logger

  The logger object.

• loglevel

  The level of the logging object.

• logfile

  The name of the logfile.

• format

  The log format string.

• colorize

  Specify if log messages are colored or not.

after_setup_task_logger

Sent after the setup of every single task logger. Used to augment logging configuration.

Provides arguments:

• logger

  The logger object.

• loglevel

  The level of the logging object.

• logfile

  The name of the logfile.

• format

  The log format string.

• colorize

  Specify if log messages are colored or not.

Command signals

user_preload_options

This signal is sent after any of the Celery command line programs are finished parsing the user preload options.

It can be used to add additional command-line arguments to the celery umbrella command:

from celery import Celery
from celery import signals
from celery.bin.base import Option

app = Celery()
app.user_options['preload'].add(Option(
    '--monitoring', action='store_true',
    help='Enable our external monitoring utility, blahblah',
))

@signals.user_preload_options.connect
def handle_preload_options(options, **kwargs):
    if options['monitoring']:
        enable_monitoring()

Sender is the Command instance, which depends on what program was called (e.g. for the umbrella command it will be a CeleryCommand) object).

Provides arguments:

• app

  The app instance.

• options

  Mapping of the parsed user preload options (with default values).

Deprecated Signals

task_sent

This signal is deprecated, please use after_task_publish instead.