celery get 和 join 的区别

Question

两者有什么区别：

 r = group(some_task.s(i) for i in range(10)).apply_async()
 result = r.join()

还有：

 r = group(some_task.s(i) for i in range(10))()
 result = r.get()

Celery 文档同时使用了这两个示例，我看不出有什么区别。

Answer 1

简答

虽然group 的get 和join 方法应该返回相同的结果，但get 实现了一些缓存，并且可能会更有效，具体取决于您使用的后端。除非你真的需要在某些极端情况下使用join，否则你应该使用get。

长答案

这里是 celery 的 ResultSet 类的 get 方法的源代码，GroupResult 类扩展了该类。

def get(self, timeout=None, propagate=True, interval=0.5,
        callback=None, no_ack=True, on_message=None):
    """See :meth:`join`
    This is here for API compatibility with :class:`AsyncResult`,
    in addition it uses :meth:`join_native` if available for the
    current result backend.
    """
    if self._cache is not None:
        return self._cache
    return (self.join_native if self.supports_native_join else self.join)(
        timeout=timeout, propagate=propagate,
        interval=interval, callback=callback, no_ack=no_ack,
        on_message=on_message,
    )

我们看到的第一件事是文档字符串告诉我们查看join 方法以获取文档。马上，这表明这些方法非常相似。

查看get 方法的主体，我们可以看到它首先检查缓存值，如果已设置则返回该值。如果没有找到缓存值，get 将调用join 或join_native 方法，具体取决于后端是否支持本机连接。如果你觉得 return 语句的格式有点混乱，这本质上是一样的：

if self.supports_native_join:
    return self.join_native(timeout=timeout,
                            propagate=propagate,
                            interval=interval,
                            callback=callback,
                            no_ack=no_ack,
                            on_message=on_message)
else:
    return self.join(timeout=timeout,
                     propagate=propagate,
                     interval=interval,
                     callback=callback,
                     no_ack=no_ack,
                     on_message=on_message)

join 方法的文档字符串是这样说的。

对于结果存储后端而言，这可能是一项代价高昂的操作，必须 诉诸轮询（例如，数据库）。你应该考虑使用 join_native 如果你的后端支持的话。

因此，如果您的后端支持，您应该调用join_native 而不是join。但是，如果get 为您包装了这个逻辑，为什么还要有条件地调用其中一个或另一个呢？只需改用get。

Answer 2

区别在于组和和弦之间的区别。问题是你是想要所有任务的结果，还是想要一个对结果做某事的任务。

组用于启动多个任务，然后按照调用顺序加入结果。

>>> job = group([
...             add.subtask((2, 2)),
...             add.subtask((4, 4)),
...             add.subtask((8, 8)),
...             add.subtask((16, 16)),
...             add.subtask((32, 32)),
... ])
>>> result = job.apply_async()
>>> result.join()
[4, 8, 16, 32, 64]

Chords 是指您想要在所有指定任务完成后执行的任务。

>>> callback = last_task.subtask()
>>> tasks = [task.subtask(...) ... ]
>>> result = chord(tasks)(callback)
>>> result.get()
<output from last_task which have access to the results from the tasks>

您可以在此处了解更多信息：http://ask.github.io/celery/userguide/tasksets.html