如何在Django中使用uWSGI Spooler替代Celery运行并发任务

问题根因 #

你遇到的unable to find the spooler function报错核心原因是uWSGI Spooler进程没有成功加载到被@spool装饰的任务函数，结合你的配置还有几个隐藏的配置错误，按下面步骤修改即可：

修复步骤 #

1. 修正uWSGI配置项
   你开启了lazy-apps = true，该配置会让所有子进程（包括工作进程、Spooler进程）在fork完成后才独立加载应用代码，ini中配置的普通import只会在master进程执行，不会被Spooler进程继承。修改uwsgi.ini的Spooling相关配置：

# Spooling
spooler             = /path/to/tasks
spooler-harakiri    = 600
# 替换原来的import，专门指定Spooler进程启动时导入的模块
spooler-import      = djproj.spooler_init
# 如果不需要lazy-apps也可以直接关闭，同时保留原来的import配置也能生效
# lazy-apps = false

2. 新增Django环境初始化脚本
   在djproj目录下新建spooler_init.py，确保导入任务前先初始化Django环境，否则直接导入tasks会触发Django未初始化报错：

import os
import django
# 替换为你实际的Django配置模块路径
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djproj.settings")
django.setup()
# 初始化完成后导入任务模块，确保Spooler进程能注册到spool函数
import djproj.tasks

3. 修正任务参数不匹配问题
   你的模型层调用时传入的参数是task_id，但任务函数里读取的是obj_id，参数名不一致会触发KeyError：

# models.py中修改参数名
def start_run_in_spooler(self):
    args = prepare_spooler_args(obj_id=self.id)
    run_task(args)

4. 给spool装饰器显式指定函数名（可选，避免隐式命名匹配失败）
   修改tasks.py中的装饰器写法：

@spool(name="run_task")
def run_task(arguments):
    # 原有逻辑不变

验证方式 #

启动uWSGI后先查看启动日志，确认存在Imported spool successfully.日志输出，再触发任务调用即可看到Running in spool.的日志，任务正常执行。

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