Celery 维护所有任务的注册表。这是工作人员在收到任务消息时如何按名称查找任务的方式:
from celery import current_app
all_task_names = current_app.tasks.keys()
all_tasks = current_app.tasks.values()
foo_task = current_app.tasks['tasks.foo']
all_task_classes = [type(task) for task in current_app.tasks.itervalues()]
仅在导入包含任务的模块时才会填充任务注册表。如果你还没有导入所有模块,你可以像 celery worker 一样导入所有配置的任务模块源:
current_app.loader.import_default_modules()
注意:import_default_modules
在 Celery 2.5 之前不存在,那时你必须这样做:
from celery.loaders import current_loader
current_loader().init_worker()
注意 2:您确定要像这样编辑所有任务类吗?Celery 提供了多种配置任务类的方法,例如:
CELERY_ANNOTATIONS = {
'*': {
'loglevel': logging.DEBUG,
'logger': logging.getLogger('foo'),
},
}
将设置所有任务的 'loglevel' 和 'logger' 属性。或者,您可以使用通用基类:
class MyTask(Task):
abstract = True # means this base task won't be registered
loglevel = logging.DEBUG
logger = logging.getLogger('foo'),