尝试使用 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 存储在组中并查询每个？

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

我正在使用 RabbitMQ 在 Windows 上的 Python 2.7.10 上运行 Celery。

Answer 1

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