当前位置：文江博客话题详情

测试 celery 任务是否仍在处理中

发布于 2024-10-05 14:33:10 字数 282 浏览 12 评论 0原文

如何测试 celery 中是否仍在处理任务 (task_id)？我有以下场景：

在 Django 视图中启动任务
将BaseAsyncResult 存储在会话中
关闭 celery 守护进程（硬），以便不再处理任务
检查任务是否“已死”

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

原文

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

回眸一笑 2024-10-12 14:33:11

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

# 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

I think there is a better way than to store a task object in the model. For example, in case you wanted to check if a group of task (parallel) have completed:

# 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()

Then in your view

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

回复收藏 0 原文

聆听风音 2024-10-12 14:33:10

在模型中定义一个字段 (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 模型中。

define a field (PickledObjectField) in your model to store the celery task:

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

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

In case your task is not specific on any model you should create one specifically for the celery tasks.

or else I suggest using django-celery. It has a nice monitoring feature:
http://ask.github.com/celery/userguide/monitoring.html#django-admin-monitor, saves the tasks details in a django model in a nice graphical way.

回复收藏 0 原文

~没有更多了~