Django中如何更新Celery Beat进程内对象的数据库属性

你当前的代码里，tasks = Task.objects.all()是作为全局变量在Celery Beat启动时执行的，这个QuerySet只会在首次求值时从数据库拉取数据并缓存结果。之后仪表盘修改数据库中的on_status，Celery Beat进程里的Task实例还是启动时的旧数据，不会自动同步。

没有命令能直接更新Celery Beat进程内的实例属性，最可靠的解决方式是每次任务执行时重新从数据库拉取最新数据，具体修改如下：

• 将查询操作移到run_tasks任务函数内部，确保每次任务触发时都读取最新的数据库记录：

app = Celery("app")

app.conf.beat_schedule = {
    "run_task": {
        "task": "run_tasks",
        "schedule": 5.0,
        "options": {
            "expires": 6.0,
        },
    },
}

# 移除全局的tasks变量

@app.task
def run_tasks():
    # 每次执行任务时重新查询数据库，获取最新的Task实例
    tasks = Task.objects.all()
    for task in tasks:
        task.run()

• 如果任务数量较多，担心查询性能，可以针对性优化查询，比如只筛选on_status=True的任务（减少后续判断）：

@app.task
def run_tasks():
    # 只查询开启状态的任务，减少循环次数
    tasks = Task.objects.filter(on_status=True)
    for task in tasks:
        task.run()

这样修改后，每次Celery Beat触发run_tasks任务时，都会从数据库读取最新的on_status值，仪表盘的修改就能实时生效了。

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