芹菜获取已注册的任务列表

Question

有没有办法获得已注册任务的列表？

我试过了:

celery_app.tasks.keys()

这只返回内置Celery任务,如celery.chord,celery.chain等.

Answer 1

from celery.task.control import  inspect
i = inspect()
i.registered_tasks()

这将给出所有工人和相关注册任务的字典.

from itertools import chain
set(chain.from_iterable( i.registered_tasks().values() ))

如果您有多个工作人员运行相同的任务,或者您只需要一组所有已注册的任务,那么它就可以完成工作.

替代方式:

从终端,您可以使用此命令获取已注册任务的转储

celery inspect registered

Answer 2

使用较新版本的 celery（4.0 及更高版本），以下似乎是正确的方法：

 from celery import current_app

 current_app.loader.import_default_modules()

 tasks = list(sorted(name for name in current_app.tasks
                            if not name.startswith('celery.')))
 return tasks

Answer 3

在 shell 中，尝试：

from celery import current_app 
print(current_app.tasks.keys())

current_app.tasks将所有任务作为字典提供。键是您正在运行的当前 celery 应用程序中所有已注册任务的名称。