Celery 源码解析八:State 和 Result

在前面几篇解析中,我们已经看过了 Worker 是如何运行的,Task 是如何创建的,以及怎么被路由到 Worker 中,除了这些之外,我们还对流量限制,Worker 控制和 Task/Worker 产生和处理 Event 进行了介绍。但这却不是全部,今天我将继续和大家一起来看看 Celery 的 Task Result 和 State 相关的内容。

Task State

celery/celery/states.py 中,可以看到 Task 的所有 State

Celery 源码解析八:State 和 Result

至于如果想要看 Task 的状态转变,那我们就需要回顾一下 Task 从产生到完成的过程,如果是一个普通的 Task 的话,那么它的执行流程应该是:

从 Producer 产生 -①-> MQ -②-> Consumer 接收 -③-> 执行策略执行 -④-> 执行完成

那么这里的 ①②③④ 有没有设置状态,并且状态是啥,就是我们关注的重点了,ok,那我们就跟随之前的脚步快速得发掘一番:

Celery 源码解析八:State 和 Result

这个是同步发送消息的时候,其实也是第一次出现 state 的时候,但是这个 state 已经是 ④ 之后的状态了,但是,无妨,我们已经知道了执行完成之后的状态是在这里 update 的。那么对于异步的情况又是如何更新状态的呢,那根据我们之前的 review,我们知道应该去找 AsyncResult:

Celery 源码解析八:State 和 Result

所以这里就是 backend 的事情啦,backend 再甩锅给 backend Consumer,然后就是在 backend Consumer 里面干活了,最后就到了这:

Celery 源码解析八:State 和 Result

这里其实虽然是调用的 on_state_change 但是,我们应该知道的一点就是在这个时候其实状态已经变化了,那么我们就应该追述到上面的 result._iter_meta,看看里面的实现,这里我就飘过很多细节了,直接看最实际的地方:

Celery 源码解析八:State 和 Result

这里可以发现,其实结果是通过 Backend 组件完成的,例如 Redis 就放在 Key 对应的 Value 中,其他的 Backend 也类似,如果没有的话,在 Line 667 我们就知道了,状态就是 PENDING 啦,第二个状态捕获!然后还有 Line 668 这里的解析结果也是值得我们一看,我直接看最后吧:

Celery 源码解析八:State 和 Result

可以发现这里也是很有趣的,如果 Task 的状态是异常状态的话,还需要进行异常回溯,其实就是把异常信息解压出来,展示给我们看。

那我们是时候我们去找找第三个状态:Received 了,我们可以猜测一下大概可能出现的位置,这里就是:

Celery 源码解析八:State 和 Result

这里可以发现居然是有一个 EVENT 发送出去,然后我们就应该去看看处理这个 Event 的人啦,如果你还有印象的话,那么你应该可以很容易得追踪到:

Celery 源码解析八:State 和 Result

这里你会发现你感兴趣的所有东西,哈哈,其实你需要看看一个字典:

Celery 源码解析八:State 和 Result

你会发现所有的状态都可以在这里更新,哈哈。所以下面我就列一下各个 EVENT 的发出地点就好了:

Celery 源码解析八:State 和 Result

如果你够细心的话,你会发现 failedsucceeded 都是有两处发送地点,有兴趣猜测一下原因么?

Task Result

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/wppjpf.html