python之Celery 任务立即自动发现

我正在尝试在 Celery 框架之上构建应用程序。

我有一个模块 settings/celery_settings.py，其中包含像这样初始化 Celery 应用程序的代码(我扩展了一些变量):

from __future__ import absolute_import 
from celery import Celery 
 
pfiles = ['other_tasks.test123', 'balance_log.balance_log'] 
app = Celery('myapp') 
# here I just have some parameters defined like broker, result backend, etc 
# app.config_from_object(settings) 
 
# TRYING to discover tasks 
app.autodiscover_tasks(pfiles) 

文件 other_tasks/test123.py 和 balance_log/balance_log.py 包含如下任务定义:

# file other_tasks/test123.py 
from celery import shared_task, Task 
 
@shared_task() 
def mytask(): 
    print("Test 1234!") 
 
class TestTask01(Task): 
 
    def run(self, client_id=None): 
        logger.debug("TestTask01: run") 
        return client_id 

我运行 celery worker :

python3 /usr/local/bin/celery -A settings.celery_settings worker 

这样它就可以发现任务。我可以调用这些任务。

但后来我尝试使用 IPython:

In [1]: from settings.celery_settings import app 
 
In [2]: app.tasks 
Out[2]:  
{'celery.backend_cleanup': <@task: celery.backend_cleanup of XExchange:0x7f9f50267ac8>, 
 'celery.chain': <@task: celery.chain of XExchange:0x7f9f50267ac8>, 
 'celery.chord': <@task: celery.chord of XExchange:0x7f9f50267ac8>, 
 'celery.chord_unlock': <@task: celery.chord_unlock of XExchange:0x7f9f50267ac8>, 
 'celery.chunks': <@task: celery.chunks of XExchange:0x7f9f50267ac8>, 
 'celery.group': <@task: celery.group of XExchange:0x7f9f50267ac8>, 
 'celery.map': <@task: celery.map of XExchange:0x7f9f50267ac8>, 
 'celery.starmap': <@task: celery.starmap of XExchange:0x7f9f50267ac8>} 

显然它没有发现任务。

似乎当我显式调用任务时，我首先导入它们并在调用时为 celery 指定确切路径。这就是它起作用的原因。

问题:如何让他们发现有已知任务列表？

请您参考如下方法:

最后我发现autodiscover_tasks函数还有一个额外的参数:

def autodiscover_tasks(self, packages, related_name='tasks', force=False): 
    ... 

因此，在设置 force=True 后它开始工作了!

app.autodiscover_tasks(pfiles, force=True)