taskflow 使用指南

赞

原来超过一定时间不能编辑的....

2. Executor

前面说了,Engine 会通过 Executor 执行任务,因为如果 Engine 直接执行任务的话,整个状态机的循环会受到正在执行的任务的影响,所以包了一层 Executor 来执行具体的任务(当然具体代码里对 Executor 的应用会更复杂一点,为了扩展和异常处理包了 3 层)

在 taskflow 的源代码中 Executor 是通过 futurist 库来实现的,而 futurist 又是基于 futures 的,这个库内部实现还是比较复杂的,如果没用过对应库的,建议直接参考我简化的[futurist]( https://github.com/lolizeppelin/simpleutil/blob/master/simpleutil/utils/futurist.py),因为是用 eventlet 实现的,所以需要熟悉 eventlet.

具体的任务代码（比如备份数据库什么的）在一般情况下可以不处理异常,因为执行任务的代码通过 except Exception 捕获了任务的所有异常.

特殊异常就是 CancelledError,这个异常是调度到已经取消任务时由 futurist 抛出,在读代码的时候需注意下这个的特殊处理

3. Scheduler

这个没什么好说的,Executor 的封装的最上层,最后执行会落实到具体的 Executor 上

4. storage

这个是存储接口,后面说到 flow 的时候会详细讲到,storage 的初始化在 Engine 中,一个功能是数据存储的接口,一个功能是作为 flow 的外层封装

4. Runtime 与 machine

在看这个之前,如果你还不熟悉状态机,建议先拿前面说的 automaton 练练手,如果已经熟悉状态机但是还没看过 automaton 代码的，建议去看看 automaton 的代码

machine 就是 Engine 中循环的(automaton)状态机了,一个 engine 只运行一个状态机,初始化代码在 builder.MachineBuilder,MachineBuilder 又是在 Runtime 中调用生成 machine 的,我们先别管 Runtime,先理解一下 taskflow 的状态机

taskflow 状态机并不复杂,但还不熟悉 taskflow 的时候很容易被高懵.因为 taskflow 用到 networkx 这个图库,而状态机其实就是一个有向图,所以一开始看的时候,很容易以为 taskflow 的状态机会非常复杂要看懂图的相关代码才能搞明白,但实际情况是

taskflow 的状态机和图无关!因为 taskflow 状态机的状态很少不需要用图来解决状态循环

那么 taskflow 为什么要用到图库呢,在解决这个疑问前我们先抛开 taskflow,自己用状态机设计一个解决前面——"停服 备份数据库 升级数据库 升级应用 启动服务器" 的工作流

1. 首先定义停服状态和对应停服状态执行的代码
2. 定义停服成功的返回,失败的返回,定义进入停服状态的 event （这个是起始时间,event 就是 start ）
3. 定义备份数据库状态对应备份执行代码
4. 定义进入备份状态的 event （前面的停服成功）
5. 定义备份成功和备份失败的返回,到目前还简单,备份失败大不了多备份几次直到成功,失败了整个状态机终止都可以影响不大
6. 定义升级数据库状态对应备份执行代码
7. 定义进入升级状态的 event （前面的备份成功）
8. 定义升级成功和备份失败的返回,这里开始坑了,升级失败要回滚了
9. 发现少了回滚升级失败的状态定义.....增加升级失败回滚失败的定义
......
回滚升级数据库失败....升级应用是失败...回滚升级应用是失败.....启动失败

设计下来你发现没几个步骤。要定义的状态就越来越多...这也就是状态机复杂以后和图有关的原因了

taskflow 非常巧妙的避免了复杂化状态机,taskflow 的设计的状态机可以简单的理解只处理 2 个状态就好

开始....找到任务-执行任务-找到任务....执行任务...终止

执行任务就是调用 Executor, 至于找下一个任务的工作,就是封装了图库的 flow 的工作了.这样设计状态机状态就很少,具体的状态可以看 MachineBuilder 的注释中有对应表格,对应状态目前粗看一下即可,知道哪个状态是找任务、哪个状态执行任务就可,有些状态涉要看了后面的 retry 相关才比较好理解,至于 flow,这个我们在后面说明

回头来看 Runtime,MachineBuilder 是由 Runtime 生成的,状态机的有些 callback 最终执行的 Runtime 中的函数,里面会有一些嵌套和封装, Scheduler 的封装就在 Runtime 中,Runtime 可以简单理解为状态机调用其它注入 Scheduler、storage 接口调用的中间件,Runtime 在整体理解 taskflow 的的时候可以不用细看

第一篇完...请看下一篇介绍 flow atom task retry

好别扭....有点文字问题都不能修正....不想发了....

赞，可以发到自己的博客

class CallJoe(task.Task):
def execute(self, joe_number, *args, **kwargs):
print("CallJoe args", args)
print("CallJoe kwargs", kwargs)
print("Calling joe %s." % joe_number)
这种定义的类, 想实现类似:
t = CallJoe()
t.execute('13911111111', "foo", "bar", baz="value_baz", qux="value_qux")
找了好久也没摸索出怎么把这个 非关键字变长参数传进去. 还请楼主指教.

不行

为什么非要变长呢 打包到 list 或者 dict 里不好嘛

具体你看反射出参数的部分怎么实现的就知道为什么不行了