报告产生了长期运行的 Celery 任务的结果 [英] Reporting yielded results of long-running Celery task

问题描述

我已将长时间运行的任务划分为逻辑子任务，因此我可以在每个子任务完成时报告结果.但是，我正在尝试报告一项实际上永远不会完成的任务的结果(而是在执行过程中产生价值)，并且我正在努力使用我现有的解决方案来做到这一点.

I've segmented a long-running task into logical subtasks, so I can report the results of each subtask as it completes. However, I'm trying to report the results of a task that will effectively never complete (instead yielding values as it goes), and am struggling to do so with my existing solution.

我正在为我编写的一些 Python 程序构建一个 Web 界面.用户可以通过网络表单提交作业，然后返回查看作业进度.

I'm building a web interface to some Python programs I've written. Users can submit jobs through web forms, then check back to see the job's progress.

假设我有两个函数，每个函数都通过单独的表单访问:

Let's say I have two functions, each accessed via separate forms:

• med_func:执行大约需要 1 分钟，结果将传递给 render()，后者会产生额外的数据.
• long_func:返回一个生成器.每个 yield 大约需要 30 分钟，并且应该报告给用户.有这么多的产量，我们可以认为这个迭代器是无限的(仅在 撤销).

• med_func: Takes ~1 minute to execute, results are passed off to render(), which produces additional data.
• long_func: Returns a generator. Each yield takes on the order of 30 minutes, and should be reported to the user. There are so many yields, we can consider this iterator as infinite (terminating only when revoked).

使用med_func，我报告结果如下:

With med_func, I report results as follows:

在提交表单时，我将 AsyncResult 保存到 Django 会话:

On form submission, I save an AsyncResult to a Django session:

    task_result = med_func.apply_async([form], link=render.s())
    request.session["task_result"] = task_result

结果页面的 Django 视图访问此 AsyncResult.任务完成后，结果将保存到一个对象中，该对象作为上下文传递给 Django 模板.

The Django view for the results page accesses this AsyncResult. When a task has completed, results are saved into an object that is passed as context to a Django template.

def results(request):
    """ Serve (possibly incomplete) results of a session's latest run. """
    session = request.session

    try:  # Load most recent task
        task_result = session["task_result"]
    except KeyError:  # Already cleared, or doesn't exist
        if "results" not in session:
            session["status"] = "No job submitted"
    else:  # Extract data from Asynchronous Tasks
        session["status"] = task_result.status
        if task_result.ready():
            session["results"] = task_result.get()
            render_task = task_result.children[0]

            # Decorate with rendering results
            session["render_status"] = render_task.status
            if render_task.ready():
                session["results"].render_output = render_task.get()
                del(request.session["task_result"])  # Don't need any more

    return render_to_response('results.html', request.session)

此解决方案仅在函数实际终止时有效.我无法将 long_func 的逻辑子任务链接在一起，因为有未知数量的 yield(long_func 循环的每次迭代可能不产生结果).

This solution only works when the function actually terminates. I can't chain together logical subtasks of long_func, because there are an unknown number of yields (each iteration of long_func's loop may not produce a result).

是否有任何明智的方法可以访问从极其长时间运行的 Celery 任务中产生的对象，以便在生成器耗尽之前显示它们?

Is there any sensible way to access yielded objects from an extremely long-running Celery task, so that they can be displayed before the generator is exhausted?

推荐答案

为了让 Celery 知道任务的当前状态，它会在你拥有的任何结果后端中设置一些元数据.您可以利用它来存储其他类型的元数据.

In order for Celery to know what the current state of the task is, it sets some metadata in whatever result backend you have. You can piggy-back on that to store other kinds of metadata.

def yielder():
    for i in range(2**100):
        yield i

@task
def report_progress():
    for progress in yielder():
        # set current progress on the task
        report_progress.backend.mark_as_started(
            report_progress.request.id,
            progress=progress)

def view_function(request):
    task_id = request.session['task_id']
    task = AsyncResult(task_id)
    progress = task.info['progress']
    # do something with your current progress

我不会在那里扔吨的数据，但它可以很好地跟踪长时间运行的任务的进度.

I wouldn't throw a ton of data in there, but it works well for tracking the progress of a long-running task.

这篇关于报告产生了长期运行的 Celery 任务的结果的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！