如何将Celery的shared_task函数拆分到独立模块中

1. 新建独立任务模块 #

在项目对应应用目录下新建tasks.py文件，把所有@shared_task装饰的异步任务全部迁移到这个文件中，补全任务依赖的导入：

from celery import shared_task
# 导入任务逻辑需要的其他依赖，比如模型、工具函数等

@shared_task
def task_to_do(params):
    return 'something'

2. 调整原业务代码的导入 #

原调用任务的业务代码不需要修改调用逻辑，只需要新增对任务模块的导入即可：

import uuid
# 从刚才新建的tasks模块导入对应的任务函数，路径根据你自己的项目结构调整
from .tasks import task_to_do

class myTask():
    def xyz():
        # 原有调用逻辑完全不需要改动
        np = task_to_do.signature((params))
        np.apply_async(task_id=uuid.uuid4())

3. 配置Celery自动发现任务 #

在Celery实例初始化文件（通常为项目根目录下的celery.py）中配置自动任务扫描，确保Celery启动时能识别到你拆分出去的任务：

from celery import Celery

# 初始化Celery实例，替换成你自己的项目名
app = Celery("my_project")

# 配置自动扫描任务，默认会扫描所有注册应用下的tasks.py文件
app.autodiscover_tasks()

# 如果你的tasks.py不在标准应用目录下，可以手动指定扫描路径
# app.autodiscover_tasks(["common.tasks", "user.tasks"])

注意事项 #

• 启动Celery Worker时要保证任务模块所在的目录在Python的模块搜索路径中，否则会出现task not registered错误
• 任务代码修改后必须重启Celery Worker才能生效
• 避免循环导入：不要在任务模块中导入原业务文件的内容，同时原业务文件导入任务模块时不要出现交叉导入的情况

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