在 Django-Q2 中，Schedule 无法直接绑定类方法（如 self.run_function），因其需序列化到数据库并由独立进程反序列化执行；必须改用字符串路径引用任务函数，并通过 kwargs 传递实例上下文。
在 django-q2 中，schedule 无法直接绑定类方法（如 `self.run_function`），因其需序列化到数据库并由独立进程反序列化执行；必须改用字符串路径引用任务函数，并通过 `kwargs` 上传递实例上下文。
Django-Q2 的 Schedule 与 async_task 在函数引用机制上存在本质差异：async_task 运行在当前请求上下文中，可直接传入已绑定的实例方法（如 <bound method TestApp.run_function of ...>）；而 Schedule 则需将任务持久化至数据库，并由独立 worker 进程（可能跨进程/跨机器）反序列化执行——此时绑定方法（bound method）因依赖特定对象实例，无法被安全序列化和重建，导致报错 Function ... is not defined 和 malformed return hook。
✅ 正确做法是：将业务逻辑提取为模块级纯函数，通过字符串路径（dotted string）注册到 Schedule，并利用 kwargs 传递关键标识（如模型 ID）以重建运行时上下文。
✅ 推荐实现结构
- 在 test_app/tasks.py 中定义无状态任务函数（注意：必须位于可导入路径下，且不能嵌套或依赖闭包）：
# test_app/tasks.py
from django.apps import apps
import ast
def run_function(**kwargs):
"""
执行指定 TestApp 实例关联的外部函数
"""
TestApp = apps.get_model('test_app', 'TestApp')
test_app_id = kwargs.get('TestApp_id')
if not test_app_id:
raise ValueError("Missing 'TestApp_id' in kwargs")
try:
test_app = TestApp.objects.get(pk=test_app_id)
except TestApp.DoesNotExist:
raise ValueError(f"TestApp with id {test_app_id} not found")
# 动态导入模块并调用函数
module = __import__(test_app.test_function.library_name)
func = getattr(module, test_app.test_function.function_name)
result = func(*test_app.get_args())
print(f"[Task] Executed {test_app.id}: {result}")
return {
'result': result,
'TestApp_id': test_app_id
}
def print_task(task):
"""
Hook 函数：接收已完成的 Task 对象
task.result 是 run_function 的返回值（dict）
"""
try:
data = task.result
app_id = data.get('TestApp_id')
res = data.get('result')
print(f"[Hook] TestApp {app_id} completed → {res}")
except (AttributeError, TypeError, KeyError) as e:
print(f"[Hook Error] Failed to process task result: {e}")- 修改模型 save() 方法，使用字符串路径创建 Schedule：
# test_app/models.py
from django.db import models
from django_q.models import Schedule
class TestApp(models.Model):
cron = models.CharField(max_length=200)
args = models.CharField(max_length=200) # 建议改用 JSONField 存储列表/字典更安全
test_function = models.ForeignKey('TestFunction', on_delete=models.CASCADE)
scheduled_task = models.ForeignKey(
Schedule,
blank=True,
null=True,
on_delete=models.SET_NULL,
related_name='triggered_testapps'
)
def get_args(self):
try:
return ast.literal_eval(self.args)
except (ValueError, SyntaxError):
return []
def save(self, *args, **kwargs):
# 避免重复创建 schedule（如更新时）
if not self.pk or not self.scheduled_task:
self.scheduled_task = Schedule.objects.create(
func='test_app.tasks.run_function', # ✅ 必须是 dotted string
hook='test_app.tasks.print_task', # ✅ 同样必须是 dotted string
schedule_type=Schedule.CRON,
cron=self.cron,
kwargs={'TestApp_id': self.pk}, # ✅ 仅传必要、可序列化的数据
repeats=-1, # 永久重复（按 cron 触发）
queue='default'
)
super().save(*args, **kwargs)⚠️ 关键注意事项
- 不可序列化内容禁止传入 kwargs：self、request、数据库连接、文件句柄等均不可序列化。只传 id、pk 等基础类型。
- 函数路径必须绝对且可导入：确保 'test_app.tasks.run_function' 在所有 worker 进程中均可通过 importlib.import_module() 加载（即 test_app 是已注册的 Django app，tasks.py 在 Python path 中）。
- 异常处理至关重要：worker 进程无调试上下文，务必在 run_function 和 print_task 中添加完整 try/catch，并记录日志（推荐使用 logging 而非 print）。
- 避免竞态条件：若 TestApp 实例可能被删除，print_task 中应捕获 DoesNotExist；建议在 run_function 中对 test_app.test_function 做存在性校验。
- 性能优化建议：args 字段建议升级为 JSONField（Django 3.1+）或 TextField + json.dumps/loads，比 ast.literal_eval 更安全高效。
通过该模式，你既保留了面向对象建模的清晰性，又满足了分布式任务调度对函数可序列化、无状态的核心要求。
