尝试使用 ID 和 AMQP 后端检索 Celery GroupResult

Question

在将 Celery 与 RabbitMQ 结合使用时，我运行遇到了一个非常棘手的问题。

我有类似的东西：

from celery import group
group_of_tasks = group(task.s(x) for x in job_list)
result = group_of_tasks.delay()
print result.id  # let's call this d453359d...

以上工作正常，没有问题，我可以查询组的状态以及 result.results 中的单个 AsyncResults。

但是，如果我尝试在单独的控制台中执行以下操作：

from celery.result import GroupResult
x = GroupResult.restore('d453359d...')

我收到以下错误：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "C:\Python27\lib\site-packages\celery\result.py", line 806, in restore
     ).restore_group(id)
  File "C:\Python27\lib\site-packages\celery\backends\amqp.py", line 297, in restore_group
    'restore_group is not supported by this backend.')
NotImplementedError: restore_group is not supported by this backend.

我在 Retrieving GroupResult from taskset_id in Celery? 中遇到了一个类似的问题，其中提到了 .save，但使用它也会导致抛出 NotImplementedError。

celery.backends.amqp 中定义的 AMQP 后端的源代码如下：

def save_group(self, group_id, result):
    raise NotImplementedError(
        'save_group is not supported by this backend.')

def restore_group(self, group_id, cache=True):
    raise NotImplementedError(
        'restore_group is not supported by this backend.')

所以，我的问题是，我是否无法仅使用 GroupResult ID 重新创建一组结果？唯一的方法是将每个 AsyncResults 的 ID 存储在组中并查询每个？

或者我只是遗漏了一些非常明显的东西？

我是运行 Celery Python 2.7.10 Windows，使用 RabbitMQ。

Answer 1

您是对的 - 使用 RabbitMQ 时无法重新创建结果组 - 您将需要使用支持此操作的不同结果后端，例如 Redis。

尝试使用 ID 和 AMQP 后端检索 Celery GroupResult

Attempting to retrieve a Celery GroupResult using ID, with an AMQP backend

python

celery

celery-task