python - 测试是否仍在处理 celery 任务

Question

如何测试任务（task_id）是否仍在celery中处理？我有以下情况：

1. 在 Django 视图中启动任务
2. 将 BaseAsyncResult 存储在会话中
3. 关闭 celery 守护进程（硬），因此不再处理任务
4. 检查任务是否“死亡”

有任何想法吗？是否可以查找 celery 正在处理的所有任务并检查我的任务是否仍然存在？

Answer 1

在模型中定义一个字段（PickledObjectField）来存储 celery 任务：

class YourModel(models.Model):
    .
    .
    celery_task = PickledObjectField()
    .
    .

    def task():
        self.celery_task = SubmitTask.apply_async(args = self.task_detail())
        self.save()

如果您的任务不是特定于任何模型的，您应该专门为 celery 任务创建一个。

否则我建议使用 django-celery。它有一个很好的监控功能：
http ://ask.github.com/celery/userguide/monitoring.html#django-admin-monitor ，以一种很好的图形方式将任务细节保存在 django 模型中。

Answer 2

我认为有比在模型中存储任务对象更好的方法。例如，如果您想检查一组任务（并行）是否已完成：

# in the script you launch the task
from celery import group

job = group(
    task1.s(param1, param2),
    task2.s(param3, param4)
)
result = job.apply_async()
result.save()

# save the result ID in your model
your_obj_model = YourModel.objects.get(id='1234')
your_obj_model.task_id = result.id
your_obj_model.save()

那么在你看来

from celery.result import GroupResult
# ...
task_result = GroupResult.restore(your_obj_model.task_id)
task_finished = task_result.ready()
# will be True or False