Celery源码阅读 result

一、ResultBase

Result基类，parent用于某个任务是一条任务链中的子任务时标记父任务。

class ResultBase:"""Base class for results."""#: Parent result (if part of a chain)parent = None

二、AsyncResult

继承自ResultBase的Result类，task默认使用AsyncResult作为结果类。

1. 属性


def __init__(self, id, backend=None,task_name=None,            # deprecatedapp=None, parent=None):if id is None:raise ValueError(f'AsyncResult requires valid id, not {type(id)}')self.app = app_or_default(app or self.app)self.id = idself.backend = backend or self.app.backendself.parent = parentself.on_ready = promise(self._on_fulfilled, weak=True)self._cache = Noneself._ignored = False

app：app，若未指定则使用默认app。
id：result对应的task的id。
backend：结果后端，未指定则使用app绑定的结果后端。
parent：父任务对应的Result对象。
on_ready:
_cache：用于保存result对应的元任务。
_ignored：若为Ture则忽略任务结果。

2. property

ignored，对应_ignored，若为True忽略任务结果。

@property
def ignored(self):"""If True, task result retrieval is disabled."""if hasattr(self, '_ignored'):return self._ignoredreturn False@ignored.setter
def ignored(self, value):"""Enable/disable task result retrieval."""self._ignored = value

观察get函数，可以发现当ignored为True时直接返回。

def get(self, timeout=None, propagate=True, interval=0.5,no_ack=True, follow_parents=True, callback=None, on_message=None,on_interval=None, disable_sync_subtasks=True,EXCEPTION_STATES=states.EXCEPTION_STATES,PROPAGATE_STATES=states.PROPAGATE_STATES):if self.ignored:return....

graph，返回以当前节点为根的有向无环子图。

@cached_property
def graph(self):return self.build_graph()

查看build_graph方法：

def build_graph(self, intermediate=False, formatter=None):graph = DependencyGraph(formatter=formatter or GraphFormatter(root=self.id, shape='oval'),)for parent, node in self.iterdeps(intermediate=intermediate):graph.add_arc(node)if parent:graph.add_edge(parent, node)return graph

build_graph生成了一个DependencyGraph，DependencyGraph是一个有向无环图，通过拓扑排序确定任务的处理顺序。graph的根是当前result节点，通过iterdeps遍历图，最终生成以当前节点为根的子图。

查看iterdeps方法：

def iterdeps(self, intermediate=False):stack = deque([(None, self)])while stack:parent, node = stack.popleft()yield parent, nodeif node.ready():stack.extend((node, child) for child in node.children or [])else:if not intermediate:raise IncompleteStream()

iterdeps通过层次遍历的方式遍历以当前result 节点为根的result子图。生成器中的每一项是当前节点和当前节点的父节点组成的元组。

supports_native_join，结果后端是否支持本地联接。

@property
def supports_native_join(self):return self.backend.supports_native_join

children，返回当前任务的子任务。

@property
def children(self):return self._get_task_meta().get('children')def _get_task_meta(self):if self._cache is None:return self._maybe_set_cache(self.backend.get_task_meta(self.id))return self._cachedef _maybe_set_cache(self, meta):if meta:state = meta['status']if state in states.READY_STATES:d = self._set_cache(self.backend.meta_from_decoded(meta))self.on_ready(self)return dreturn metadef _get_task_meta(self):if self._cache is None:return self._maybe_set_cache(self.backend.get_task_meta(self.id))return self._cachedef _set_cache(self, d):children = d.get('children')if children:d['children'] = [result_from_tuple(child, self.app) for child in children]self._cache = dreturn d

_get_task_meta方法首先检查是否设置了缓存，如果没有设置则调用结果后端通过task id获取task，并通过_maybe_set_cache将任务进行缓存。_set_cache调用result_from_tuple()通过当前任务的子任务获取到子任务对应的AsyncResult对象或者GroupResult对象。

三、ResultSet(后续更新)