python - 报告长期运行的 Celery 任务的结果

Question

问题

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

背景

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

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

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

代码，当前实现

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

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

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

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

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 循环的每次迭代可能不产生结果)。

问题

有没有什么明智的方法可以从一个非常长时间运行的 Celery 任务中访问产生的对象，以便在生成器耗尽之前显示它们？

Answer 1

为了让 Celery 知道任务的当前状态是什么，它会在您拥有的任何结果后端设置一些元数据。您可以搭载它来存储其他类型的元数据。

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

我不会在其中抛出 吨 数据，但它可以很好地跟踪长时间运行的任务的进度。