应用程序¶

Application

中文

在使用 Celery 库之前，必须先实例化它，这个实例被称为应用（或简称 app）。

该应用是线程安全的，因此多个具有不同配置、组件和任务的 Celery 应用可以共存于同一个进程空间中。

现在我们来创建一个：

>>> from celery import Celery
>>> app = Celery()
>>> app
<Celery __main__:0x100469fd0>

最后一行展示了该应用的文本表示形式：包括应用类的名称（ Celery ）、当前主模块的名称（ __main__ ）以及对象的内存地址（ 0x100469fd0 ）。

英文

The Celery library must be instantiated before use, this instance is called an application (or app for short).

The application is thread-safe so that multiple Celery applications with different configurations, components, and tasks can co-exist in the same process space.

Let's create one now:

>>> from celery import Celery
>>> app = Celery()
>>> app
<Celery __main__:0x100469fd0>

The last line shows the textual representation of the application: including the name of the app class (Celery), the name of the current main module (__main__), and the memory address of the object (0x100469fd0).

主名称¶

Main Name

中文

这三者中，只有一个是重要的，那就是主模块名称。我们来看一下原因。

当你在 Celery 中发送一个任务消息时，该消息不会包含任何源代码，只包含你想执行的任务名称。这类似于互联网中的主机名工作原理：每个 worker 维护着一个任务名称与其实际函数的映射，称为 任务注册表 （task registry）。

每当你定义一个任务时，该任务也会被添加到本地注册表中：

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.name
__main__.add

>>> app.tasks['__main__.add']
<@task: __main__.add>

你再次看到 __main__；当 Celery 无法检测函数所属模块时，它会使用主模块名来生成任务名称的开头部分。

这种情况只在一小部分使用场景中才会造成问题：

如果定义任务的模块作为程序运行；
如果在 Python 解释器（REPL）中创建应用；

比如下面这个例子，其中 tasks 模块也用于通过 app.worker_main() 启动一个 worker：

tasks.py:

from celery import Celery
app = Celery()

@app.task
def add(x, y): return x + y

if __name__ == '__main__':
    args = ['worker', '--loglevel=INFO']
    app.worker_main(argv=args)

当该模块被执行时，任务的名称将以 "__main__" 开头，但当该模块被另一个进程导入（例如为了调用一个任务）时，任务名称将以 "tasks" 开头（也就是模块的实际名称）：

>>> from tasks import add
>>> add.name
tasks.add

你也可以为主模块指定一个其他名称：

>>> app = Celery('tasks')
>>> app.main
'tasks'

>>> @app.task
... def add(x, y):
...     return x + y

>>> add.name
tasks.add

参见

名称

英文

Only one of these is important, and that's the main module name. Let's look at why that is.

When you send a task message in Celery, that message won't contain any source code, but only the name of the task you want to execute. This works similarly to how host names work on the internet: every worker maintains a mapping of task names to their actual functions, called the task registry.

Whenever you define a task, that task will also be added to the local registry:

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.name
__main__.add

>>> app.tasks['__main__.add']
<@task: __main__.add>

and there you see that __main__ again; whenever Celery isn't able to detect what module the function belongs to, it uses the main module name to generate the beginning of the task name.

This is only a problem in a limited set of use cases:

If the module that the task is defined in is run as a program.
If the application is created in the Python shell (REPL).

For example here, where the tasks module is also used to start a worker with app.worker_main():

tasks.py:

from celery import Celery
app = Celery()

@app.task
def add(x, y): return x + y

if __name__ == '__main__':
    args = ['worker', '--loglevel=INFO']
    app.worker_main(argv=args)

When this module is executed the tasks will be named starting with "__main__", but when the module is imported by another process, say to call a task, the tasks will be named starting with "tasks" (the real name of the module):

>>> from tasks import add
>>> add.name
tasks.add

You can specify another name for the main module:

>>> app = Celery('tasks')
>>> app.main
'tasks'

>>> @app.task
... def add(x, y):
...     return x + y

>>> add.name
tasks.add

参见

名称

惰性¶

Laziness

中文

应用实例是惰性加载的，即只有在真正需要时才会进行初始化。

创建一个 Celery 实例时只会执行以下操作：

创建逻辑时钟实例，用于事件系统。
创建任务注册表。
将自身设置为当前应用（除非禁用了 set_as_current 参数）。
调用 app.on_init() 回调（默认无操作）。

app.task() 装饰器在任务被定义时不会立即创建任务，而是将任务的创建延迟到任务被使用，或应用被 finalize 之后。

下面的示例展示了任务在实际使用前并未真正创建：

>>> @app.task
>>> def add(x, y):
...    return x + y

>>> type(add)
<class 'celery.local.PromiseProxy'>

>>> add.__evaluated__()
False

>>> add        # <-- 导致调用 repr(add)
<@task: __main__.add>

>>> add.__evaluated__()
True

应用的 finalization 可以显式通过调用 app.finalize() 方法完成，或者隐式通过访问 app.tasks 属性来触发。

完成 finalization 会执行以下操作：

复制需要在多个应用间共享的任务。

默认情况下任务是共享的，但如果在 task 装饰器中禁用了 shared 参数，那么该任务将仅绑定于其所归属的应用实例中。
执行所有待处理的任务装饰器。
确保所有任务都绑定到当前应用上。

任务之所以需要绑定到应用，是为了从配置中读取默认值等信息。

英文

The application instance is lazy, meaning it won't be evaluated until it's actually needed.

Creating a Celery instance will only do the following:

Create a logical clock instance, used for events.
Create the task registry.
Set itself as the current app (but not if the set_as_current argument was disabled)
Call the app.on_init() callback (does nothing by default).

The app.task() decorators don't create the tasks at the point when the task is defined, instead it'll defer the creation of the task to happen either when the task is used, or after the application has been finalized,

This example shows how the task isn't created until you use the task, or access an attribute (in this case repr()):

>>> @app.task
>>> def add(x, y):
...    return x + y

>>> type(add)
<class 'celery.local.PromiseProxy'>

>>> add.__evaluated__()
False

>>> add        # <-- causes repr(add) to happen
<@task: __main__.add>

>>> add.__evaluated__()
True

Finalization of the app happens either explicitly by calling app.finalize() -- or implicitly by accessing the app.tasks attribute.

Finalizing the object will:

Copy tasks that must be shared between apps

Tasks are shared by default, but if the shared argument to the task decorator is disabled, then the task will be private to the app it's bound to.
Evaluate all pending task decorators.
Make sure all tasks are bound to the current app.

Tasks are bound to an app so that they can read default values from the configuration.

打破链条¶

Breaking the chain

中文

尽管依赖当前应用被设置是可行的，但最佳实践是始终将应用实例作为参数传递给需要它的地方。

我将此称为 “应用链（app chain）” ，因为它会形成一个依赖于显式传递 app 的实例链。

以下示例被认为是不良实践：

from celery import current_app

class Scheduler:

    def run(self):
        app = current_app

正确的做法是将 app 作为参数传入：

class Scheduler:

    def __init__(self, app):
        self.app = app

Celery 内部使用 celery.app.app_or_default() 函数，以确保模块兼容性 API 中一切也能正常工作：

from celery.app import app_or_default

class Scheduler:
    def __init__(self, app=None):
        self.app = app_or_default(app)

在开发过程中，你可以设置 CELERY_TRACE_APP 环境变量，当“应用链”断裂时抛出异常：

$ CELERY_TRACE_APP=1 celery worker -l INFO

英文

While it's possible to depend on the current app being set, the best practice is to always pass the app instance around to anything that needs it.

I call this the "app chain", since it creates a chain of instances depending on the app being passed.

The following example is considered bad practice:

from celery import current_app

class Scheduler:

    def run(self):
        app = current_app

Instead it should take the app as an argument:

class Scheduler:

    def __init__(self, app):
        self.app = app

Internally Celery uses the celery.app.app_or_default() function so that everything also works in the module-based compatibility API

from celery.app import app_or_default

class Scheduler:
    def __init__(self, app=None):
        self.app = app_or_default(app)

In development you can set the CELERY_TRACE_APP environment variable to raise an exception if the app chain breaks:

$ CELERY_TRACE_APP=1 celery worker -l INFO

抽象任务¶

Abstract Tasks

中文

所有使用 app.task() 装饰器创建的任务都会继承应用的基类 Task。

你可以通过 base 参数指定不同的基类：

@app.task(base=OtherTask):
def add(x, y):
    return x + y

若要创建自定义任务类，应从中性基类 celery.Task 继承：

from celery import Task

class DebugTask(Task):

    def __call__(self, *args, **kwargs):
        print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))
        return self.run(*args, **kwargs)

小技巧

如果你重写了任务的 __call__ 方法，务必要调用 self.run 来执行任务主体。不要调用 super().__call__。中性基类 celery.Task 的 __call__ 方法仅供参考。出于优化目的，该方法在内部被展开为 celery.app.trace.build_tracer.trace_task，如果未定义 __call__ 方法，Celery 会直接在自定义任务类上调用 run。

中性基类的特殊之处在于它尚未绑定到任何特定的 app。一旦任务绑定到某个 app，它就会读取配置来设置默认值等。

要实现一个基类，需通过 app.task() 装饰器创建任务：

@app.task(base=DebugTask)
def add(x, y):
    return x + y

你甚至可以通过修改应用的 app.Task() 属性来改变应用的默认基类：

>>> from celery import Celery, Task

>>> app = Celery()

>>> class MyBaseTask(Task):
...    queue = 'hipri'

>>> app.Task = MyBaseTask
>>> app.Task
<unbound MyBaseTask>

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.__class__.mro()
[<class add of <Celery __main__:0x1012b4410>>,
 <unbound MyBaseTask>,
 <unbound Task>,
 <type 'object'>]

英文

All tasks created using the app.task() decorator will inherit from the application's base Task class.

You can specify a different base class using the base argument:

@app.task(base=OtherTask):
def add(x, y):
    return x + y

To create a custom task class you should inherit from the neutral base class: celery.Task.

from celery import Task

class DebugTask(Task):

    def __call__(self, *args, **kwargs):
        print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))
        return self.run(*args, **kwargs)

小技巧

If you override the task's __call__ method, then it's very important that you also call self.run to execute the body of the task. Do not call super().__call__. The __call__ method of the neutral base class celery.Task is only present for reference. For optimization, this has been unrolled into celery.app.trace.build_tracer.trace_task which calls run directly on the custom task class if no __call__ method is defined.

The neutral base class is special because it's not bound to any specific app yet. Once a task is bound to an app it'll read configuration to set default values, and so on.

To realize a base class you need to create a task using the app.task() decorator:

@app.task(base=DebugTask)
def add(x, y):
    return x + y

It's even possible to change the default base class for an application by changing its app.Task() attribute:

>>> from celery import Celery, Task

>>> app = Celery()

>>> class MyBaseTask(Task):
...    queue = 'hipri'

>>> app.Task = MyBaseTask
>>> app.Task
<unbound MyBaseTask>

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.__class__.mro()
[<class add of <Celery __main__:0x1012b4410>>,
 <unbound MyBaseTask>,
 <unbound Task>,
 <type 'object'>]

应用程序¶

主名称¶

配置¶

`config_from_object`¶

示例 1：使用模块名称¶

示例 2：传递实际模块对象¶

示例 3：使用配置类/对象¶

`config_from_envvar`¶

审查配置¶

惰性¶

打破链条¶

抽象任务¶

应用程序¶

主名称¶

配置¶

config_from_object¶

示例 1：使用模块名称¶

示例 2：传递实际模块对象¶

示例 3：使用配置类/对象¶

config_from_envvar¶

审查配置¶

惰性¶

打破链条¶

抽象任务¶

`config_from_object`¶

`config_from_envvar`¶