访问 Celery 任务的部分结果答案

【问题标题】：Access partial results of a Celery task访问 Celery 任务的部分结果
【发布时间】：2016-12-25 19:14:45
【问题描述】：

我不是 Python 专家，但是我正在尝试开发一些长期运行的基于 Celery 的任务，我可以访问它们的部分结果，而不是等待任务完成。

正如您在下面的代码中看到的，给定一个乘数、一个初始和最终范围，worker 创建一个大小为 final_range - initial_range + 1 的列表。

from celery import Celery
app = Celery('trackers', backend='amqp', broker='amqp://')

@app.task
def worker(value, initial_range, final_range):
    if initial_range < final_range
        list_values = []
        for index in range(initial_frame, final_frame + 1):
            list_values.append(value * index)
        return list_values
    else
        return None

因此，与其等待所有 四个工人 完成，我想在实际返回之前访问要返回的值 (list_values)。

from trackers import worker

res_1 = worker.delay(3, 10, 10000000)
res_2 = worker.delay(5, 01, 20000000)
res_3 = worker.delay(7, 20, 50000000)
res_4 = worker.delay(9, 55, 99999999)

首先，这可能吗？如果是这样，我必须执行哪些更改才能使其正常工作？

【问题讨论】：

您的意思是要在不等待 res_4 的情况下访问 res_1，还是要在作业完成之前访问 list_values？
我想在作业完成之前访问 list_values
那么您将需要一个事实来源，您可以将值转储到其中并从中读取。如果您在单台机器上运行，这可能就像创建工作人员和父进程都可以访问的全局数据结构一样简单（尽管这通常不是一个好主意 - 获得对全局变量的多线程访问非常困难)。更好的解决方案是使用像数据库这样的外部事实来源。我已经将 redis 用于这样的事情 - 工作人员将值附加到数据结构中，父级读取它，然后 redis 处理线程和访问问题。
感谢@PaulBecotte，但我如何才能从另一个任务（同时运行）或从主方法访问（读/写）事实来源？
其实我没有考虑过——启动 celery Worker 的基本方式将进程完全分离，这使得共享全局变量变得不可能。但是，您必须为 Celery 设置传输来处理任务，因此使用它将数据传回父进程应该很容易。

标签： python celery jobs

【解决方案1】：

您绝对需要使用外部存储，例如 SQL 或 Redis/Memcached，因为通常情况下不同的任务可以在不同的服务器上执行。

因此，在您的示例中，您应该将 list_values 存储在某个数据库中并在循环期间对其进行更新。

【讨论】：

我发现 Redis 是某种主存数据库（无需访问磁盘）。因此，我宁愿使用Redis，我只是不知道如何访问worker之外的部分结果。
Redis 是这个任务的不错选择。每个工人都是一个单独的操作系统进程，所以外面的任何人都无法读取他的内存。对于 Linux 上的进程间通信，可以使用命名管道、共享内存（例如使用 tmpfs）和套接字，所有这些方法都让位于 Redis，因为它易于使用且易于扩展。

【解决方案2】：

我建议您拆分每个任务中执行的工作。

如果您需要第一个值来计算列表后面的值，那么您可以通过link 参数链接任务。见http://docs.celeryproject.org/en/latest/userguide/canvas.html#callbacks

【讨论】：