Celery任务未注册异常排查：已检测任务却报错

项目结构 #

• run.py 文件内容：

from cocapi.task import my_add

from cocapi import app
print(app.tasks.keys())

res = my_add.delay(1, 2)
print(res.get())

• cocapi/init.py 文件内容：

from celery import Celery

app = Celery('cocapi')
app.config_from_object('cocapi.celeryconfig')

• cocapi/task.py 文件内容：

from cocapi import app

@app.task
def my_add(a, b):
    return a + b

错误现象 #

运行 run.py 时抛出错误：

celery.exceptions.NotRegistered: 'cocapi.task.my_add'

但执行 print(app.tasks.keys()) 输出显示任务已存在：

dict_keys(['celery.chord_unlock', 'celery.group', 'celery.map', 'cocapi.task.my_add', 'celery.chain', 'celery.starmap', 'celery.accumulate', 'celery.chord', 'celery.backend_cleanup', 'celery.chunks'])

Worker启动命令：

celery -A cocapi worker --loglevel=info

核心原因 #

Worker进程与执行run.py的主进程任务加载逻辑不一致：

1. 启动Worker时指定入口为cocapi，但cocapi/__init__.py未导入task.py，导致Worker进程根本没加载my_add任务；
2. 执行run.py时先导入了cocapi.task.my_add，触发任务注册到主进程的app中，所以主进程能看到任务，但Worker无此任务，无法处理调用请求。

解决步骤 #

方法一：修改cocapi/__init__.py，添加任务模块导入：

from celery import Celery

app = Celery('cocapi')
app.config_from_object('cocapi.celeryconfig')
# 导入任务模块，确保Worker启动时加载任务
from . import task

方法二：在celeryconfig.py中配置任务导入：

# cocapi/celeryconfig.py
imports = ('cocapi.task',)

完成配置后，重启Worker进程，再运行run.py即可正常执行任务。

内容的提问来源于stack exchange，提问作者finlay morrison