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

Z时代
2024-01-10
分类：问答

问题

我已经将一个长期运行的任务划分为多个逻辑子任务，因此我可以在每个子任务完成时报告结果。但是，我正在尝试报告将永远无法完成的任务的结果（而不是不断产生价值），并且正在使用现有的解决方案来做到这一点。

背景

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

假设我有两个函数，每个函数都可以通过单独的形式进行访问：

med_func：执行大约需要1分钟，结果传递给render()，产生了其他数据。

long_func：返回一个生成器。每次yield大约需要30分钟，应向用户报告。产量很多，我们可以认为此迭代器是无

限的（仅在被吊销时终止）。

代码，当前实现

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

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

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

结果页面的Django视图访问this 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，因为存在未知数量的yields（每个循环的迭代long_func可能不会产生结果）。

题

有什么明智的方法可以从极其长时间运行的Celery任务访问生成的对象，以便可以在生成器用尽之前显示它们？

回答：

为了让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