Celery获取已注册任务列表

Question

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

我试过了：

celery_app.tasks.keys()

它只返回内置的Celery任务，如Celery.chord、Celery.chain等。

Answer 1

对于较新版本的芹菜（4.0或更高版本），我们可以获得如下注册任务。

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

对于较老版本的芹菜，芹菜<4、我们可以按照如下方式获取注册任务。

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

要检查与特定应用程序相关的任务，您可以传递应用程序名称

celery -A app_name inspect registered

Answer 2

对于较新版本的芹菜（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具有可用作字典的所有任务。这些键是您正在运行的当前芹菜应用程序中已注册任务的所有名称。