Celery中文翻译Application

Celery在使用前必须实例化&＃xff0c;称为application或app。app是线程安全的&＃xff0c;具有不同配置、组件、task的多个Celery应用可以在同一个进程空间共存。

# 创建Celery应用
>>> from celery import Celery
>>> app &＃61; Celery()
>>> app


最后一行文本化显示了Celery应用&＃xff1a;包含应用所属类的名称&＃xff0c;当前主模块名&＃xff0c;以及内存地址。唯一重要的信息是模块名称。

Main Name

在Celery中发送task消息时&＃xff0c;该消息仅包含要执行的task的名称。每一个worker维护一个task名称和对应函数的映射&＃xff0c;这称为task registry。

当定义一个task时&＃xff0c;该task将注册到本地&＃xff1a;

>>> &＃64;app.task
... def add(x, y):
... return x &＃43; y>>> add
<&＃64;task: __main__.add>>>> add.name
__main__.add>>> app.tasks[&＃39;__main__.add&＃39;]
<&＃64;task: __main__.add>

当Celery无法检测task函数属于哪个模块时&＃xff0c;使用main模块名生成初始task名称。

这种方式仅适用于以下两种场景&＃xff1a;

1. 定义task的模块作为程序运行
2. app在python shell中创建

# tasks.py
from celery import Celery
app &＃61; Celery()&＃64;app.task
def add(x, y): return x &＃43; yif __name__ &＃61;&＃61; &＃39;__main__&＃39;:app.worker_main()

如果直接运行tasks.py&＃xff0c;task名将以__main__为前缀&＃xff0c;但如果tasks.py被其他程序导入&＃xff0c;task名将以tasks为前缀。如下&＃xff1a;

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

也可以直接指定主模块名&＃xff1a;

>>> app &＃61; Celery(&＃39;tasks&＃39;)
>>> app.main
&＃39;tasks&＃39;>>> &＃64;app.task
... def add(x, y):
... return x &＃43; y>>> add.name
tasks.add

Configuration

可以通过直接设置&＃xff0c;或使用专用配置模块对Celery进行配置。

通过app.conf属性查看或直接设置配置&＃xff1a;

>>> app.conf.timezone
&＃39;Europe/London&＃39;>>> app.conf.enable_utc &＃61; True

或用app.conf.update方法一次更新多个配置&＃xff1a;

>>> app.conf.update(
... enable_utc&＃61;True,
... timezone&＃61;&＃39;Europe/London&＃39;,
...)

config_from_object

app.config_from_object()方法从配置模块或对象中导入配置。需要注意的是&＃xff1a;调用config_from_object()方法将重置在这之前配置的任何设置。

使用模块名
app.config_from_object()方法接收python模块的完全限定名(fully qualified name)或具体到其中的某个属性名&＃xff0c;例如"celeryconfig", "myproj.config.celery", 或"myproj.config:CeleryConfig":

from celery import Celeryapp &＃61; Celery()
app.config_from_object(&＃39;celeryconfig&＃39;)

只要能够正常执行import celeryconfig&＃xff0c;app就能正常配置。

使用模块对象
也可以传入一个已导入的模块对象&＃xff0c;但不建议这样做。

import celeryconfigfrom celery import Celeryapp &＃61; Celery()
app.config_from_object(celeryconfig)

更推荐使用模块名的方式&＃xff0c;因为这样在使用prefork pool时不需要序列化该模块。如果在实际应用中出现配置问题或序列化错误&＃xff0c;请尝试使用模块名的方式。

使用配置类或对象

from celery import Celeryapp &＃61; Celery()class Config:enable_utc &＃61; Truetimezone &＃61; &＃39;Europe/London&＃39;app.config_from_object(Config)

config_from_envvar

app.config_from_envvar()方法从环境变量中接收配置模块名。

import os
from celery import Celery#: Set default configuration module name
os.environ.setdefault(&＃39;CELERY_CONFIG_MODULE&＃39;, &＃39;celeryconfig&＃39;)app &＃61; Celery()
app.config_from_envvar(&＃39;CELERY_CONFIG_MODULE&＃39;)

通过环境变量指定配置模块&＃xff1a;

$ CELERY_CONFIG_MODULE&＃61;"celeryconfig.prod" celery worker -l info

Censored configuration

如果要显示Celery配置&＃xff0c;可能需要过滤某些敏感信息如密码、密钥等。Celery提供了几种用于帮助显示配置的实用方法。

humanize()
该方法返回列表字符串形式的配置&＃xff0c;默认只包含改动过的配置&＃xff0c;如果要显示内置的默认配置&＃xff0c;设置with_defaults参数为True&＃xff1a;

>>> app.conf.humanize(with_defaults&＃61;False, censored&＃61;True)

table()
该方法返回字典形式的配置&＃xff1a;

>>> app.conf.table(with_defaults&＃61;False, censored&＃61;True)

Celery可能不会移除所有的敏感信息&＃xff0c;因为它使用正则表达式匹配键并判断是否移除。如果用户添加了包含敏感信息的自定义配置&＃xff0c;可以使用Celery可能标记为敏感配置的名称来命名(API, TOKEN, KEY, SECRET, PASS, SIGNATURE, DATABASE)。

Laziness

应用实例是惰性的。

创建Celery实例只会执行以下操作&＃xff1a;

1. 创建用于event的logical clock instance
2. 创建task registry
3. 设置为当前应用(除非禁用了set_as_current参数)
4. 调用app.on_init()回调函数(默认不执行任何操作)

app.task()装饰器不会在task定义时立即创建task&＃xff0c;而是在task使用时或finalized应用后创建。

下例说明了在使用task或访问其属性前&＃xff0c;都不会创建task&＃xff1a;

>>> &＃64;app.task
>>> def add(x, y):
... return x &＃43; y>>> type(add)
>>> add.__evaluated__()
False>>> add # <-- causes repr(add) to happen
<&＃64;task: __main__.add>>>> add.__evaluated__()
True

应用的Finalization指显式地调用app.finalize()方法或隐式地访问app.tasks属性。

finalized应用将会&＃xff1a;

1. 复制必须在应用间共享的task。task默认是共享的&＃xff0c;但如果禁用了task装饰器的shared属性&＃xff0c;将属于应用私有。
2. 评估所有待处理的task装饰器
3. 确保所有task绑定到当前应用。将task绑定到某个应用&＃xff0c;以便可以从配置中读取默认值。

Breaking the chain

虽然可以依赖于当前应用&＃xff0c;但最佳实践是将应用实例传递给任何需要它的对象&＃xff0c;这个行为可以称为app chain。

# 依赖于当前应用(bad)from celery import current_appclass Scheduler(object):def run(self):app &＃61; current_app

# 传递应用实例(good)class Scheduler(object):def __init__(self, app):self.app &＃61; app

在开发模式设置CELERY_TRACE_APP环境变量&＃xff0c;可以在应用链断开时抛出异常&＃xff1a;

$ CELERY_TRACE_APP&＃61;1 celery worker -l info

Abstract Tasks

使用task()装饰器创建的task都继承自celery.app.task模块的Task基类。继承该类可以自定义task类&＃xff1a;

from celery import Task
# 或者 from celery.app.task import Taskclass DebugTask(Task):def __call__(self, *args, **kwargs):print(&＃39;TASK STARTING: {0.name}[{0.request.id}]&＃39;.format(self))return super(DebugTask, self).__call__(*args, **kwargs)

如果要重写__call__()方法&＃xff0c;记得调用super。这样在task直接调用时会执行基类的默认事件。

Task基类是特殊的&＃xff0c;因为它并未绑定到任何特定的应用。一旦task绑定到应用&＃xff0c;它将读取配置以设置默认值等。

1. 通过base参数指定基类

   &＃64;app.task(base&＃61;DebugTask)
def add(x, y):return x &＃43; y

2. 通过app.Task属性指定基类

   >>> from celery import Celery, Task>>> app &＃61; Celery()>>> class MyBaseTask(Task):
... queue &＃61; &＃39;hipri&＃39;>>> app.Task &＃61; MyBaseTask
>>> app.Task
>>> &＃64;app.task
... def add(x, y):
... return x &＃43; y>>> add
<&＃64;task: __main__.add>>>> add.__class__.mro()
[>,,,]