如何判断任务是否已在django-芹菜中排队？

Question

这是我的设置:

• django 1.3
• 芹菜2.2.6
• django-celery 2.2.4
• djkombu 0.9.2

在我的settings.py文件中

BROKER_BACKEND = "djkombu.transport.DatabaseTransport"

即我只是使用数据库来排队任务.

现在我的问题是:我有一个用户启动的任务,可能需要几分钟才能完成.我希望每个用户只运行一次任务,我会将任务的结果缓存在一个临时文件中,这样如果用户再次启动任务,我只需返回缓存的文件.我的视图函数中的代码如下所示:

task_id = "long-task-%d" % user_id
result = tasks.some_long_task.AsyncResult(task_id)

if result.state == celery.states.PENDING:
    # The next line makes a duplicate task if the user rapidly refreshes the page
    tasks.some_long_task.apply_async(task_id=task_id)
    return HttpResponse("Task started...")
elif result.state == celery.states.STARTED:
    return HttpResponse("Task is still running, please wait...")
elif result.state == celery.states.SUCCESS:
    if cached_file_still_exists():
        return get_cached_file()
    else:
        result.forget()
        tasks.some_long_task.apply_async(task_id=task_id)
        return HttpResponse("Task started...")

这段代码几乎可行.但是当用户快速重新加载页面时,我遇到了问题.在任务排队和最终将任务从队列中拉出并提供给工作人员之间有1-3秒的延迟.在此期间,任务的状态仍为PENDING,这会导致视图逻辑启动重复任务.

我需要的是一些方法来判断任务是否已经提交到队列中,所以我最终不会提交两次.在芹菜中有这样做的标准方法吗？

Answer 1

我用Redis解决了这个问题。只需在 Redis 中为每个任务设置一个密钥，然后在任务的 after_return 方法中从 Redis 中删除该密钥即可。Redis 轻量且快速。

Answer 2

我不认为（正如 Tomek 和其他人所建议的那样）使用数据库是实现这种锁定的方法。django 有内置的缓存框架，它应该足以完成这种锁定，并且必须更快。看：

http://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html#cookbook-task-serial

Django 可以配置为用作memcached其缓存后端，并且可以分布在多台机器上......这对我来说似乎更好。想法？

Answer 3

您可以通过手动将结果存储在数据库中来进行一些作弊。让我解释一下这会有什么帮助。

例如，如果使用 RDBMS（包含列的表 - task_id、state、result）：

查看部分：

1. 使用事务管理。
2. 使用 SELECT FOR UPDATE 获取其中 task_id == "long-task-%d" % user_id 的行。SELECT FOR UPDATE 将阻止其他请求，直到该请求提交或回滚。
3. 如果不存在 - 将状态设置为 PENDING 并启动“some_long_task”，结束请求。
4. 如果状态为 PENDING - 通知用户。
5. 如果状态为 SUCCESS - 将状态设置为 PENDING，启动任务，返回“结果”列指向的文件。我的假设是，您希望重新运行任务以获得结果。犯罪
6. 如果状态为 ERROR - 将状态设置为 PENDING，启动任务，通知用户。犯罪

任务部分：

1. 准备文件，包装在 try、catch 块中。
2. 成功时 - 用状态 = SUCCESS、结果更新正确的行。
3. 失败时 - 用状态 = ERROR 更新正确的行。