Taskflow 有兴趣了解一下？

最近在工作中会经常使用到Taskflow这个东西，看起来虽然不是很难，但是遇到各种重写的时候看起来还是有点烦的，这时候就必须来了解一下taskflow这样一个东西了。

声明：

请原谅我自己手工作的图！

TaskFlow是OpenStack开源的Python库，它帮助使任务执行变得简单、一致、可伸缩和可靠。它允许创建轻量级任务对象或函数，这些对象或函数以声明的方式组合到Flow中。它包括以一种可以停止、恢复和安全地恢复的方式运行这些Flow的引擎。使用这个库实现的项目可以享受额外的状态弹性、自然的声明式构造、更容易测试(因为任务只做一件事)、工作流可插拔性、容错和简化的崩溃恢复/容错(以及更多)。

简而言之就是：TaskFlow支持创建不同的 task，并以声明的方式集成到一个 flow 中，这些 flow 会通过 engine 执行、停止、继续和恢复。

Taskflow,先有task，再有flow，那我们先来看看task吧！
Task很简单，就是把你想执行的任务放到一个类（这个类要继承自task类）里面的execute方法里就可以了。

具体来说就是：

from taskflow import task

class TaskA(task.Task):
def execute(self,str,*args,**kwargs):
print('Hello,I am {0}'.format(str))

Task说完了，创建的Task可以是任何不同的Task，他们既可以是顺序的，也可以是毫无关联的，还可以是微相关的。为什么可以这么做呢？

因为在Flow里，一共提供三种Flow的执行方式来解决task关联性的问题。

例如，我现在要做一件事，我需要先在A表中去查到数据a,之后再依靠a作为条件去B表中查询到b，最后再把a,b处理成结果c，那么这时候就很显然看到我们有三个task，并且这三个task是顺序执行的：

这个时候我们就需要顺序执行这三个task了，我们使用线性流（linear_flow）

from taskflow.patterns import linear_flow

linear_flow.Flow('linear').add(
 taskA(),
    taskB(),
    taskC()
)

具体实现代码就是这样了。

那如果我们想做的三个task A,B,C没有依赖关系呢？那我们的三个任务是不是就完全可以并行执行？

回答：当然是！

taskflow里面又为我们提供了一种流叫无序流（unordered_flow）

它可以让这些task并行执行，就像python里面的多线程一样，谁先抢到资源谁就先执行，等到三个都执行完毕了，这个流就结束了。

from taskflow.patterns import unordered_flow

unordered_flow.Flow('linear').add(
 taskA(),
    taskB(),
    taskC()
)

最后还有一种流是图流（graph_flow），官方是这么解释的：

所包含的流/任务将根据它们的依赖关系执行，这些依赖关系将通过使用流/任务提供的和需要的映射来解决，或者通过遵循手动创建的依赖关系链接来解决。

根据依赖关系构建有向图。如果它有边A -> B，这意味着B依赖于A(并且B的执行必须等到A完成执行，而恢复意味着A的恢复必须等到B完成恢复)。

https://docs.openstack.org/taskflow/latest/user/patterns.html#module-taskflow.patterns.graph_flow

这个意思和之前的顺序是不一样的，就是说，比如我有taskA和taskB,在执行A的时候我里面可以会用到一些和B挂钩的事情，执行B的时候也同样，Task A，B之间存在依赖关系，那我们这个时候就可以使用图流了：

实现：

from taskflow.patterns import graph_flow

graph_flow.Flow('linear').add(
 taskA(),
    taskB()
)

以上就是三种流的执行方式，那么如果在flow的执行过程中，某一条task执行出现错误了，那会怎么办呢？

这个时候就task里面的revert方法就体现出了很厉害的作用了。

这里的机制其实和SQL事务里面的机制是类似的，当我在flow里面的某条task执行有问题的时候，整个flow也出现了问题。这个时候我就全部回滚，让这些task都回到一开始的初始状态。

可以看一下官方文档的例子：

Conceptual example

This pseudocode illustrates what how a flow would work for those who are familiar with SQL transactions.

START TRANSACTION task1: call nova API to launch a server || ROLLBACK task2: when task1 finished, call cinder API to attach block storage to the server || ROLLBACK ...perform other tasks...
COMMIT

The above flow could be used by Heat (for example) as part of an orchestration to add a server with block storage attached. It may launch several of these in parallel to prepare a number of identical servers (or do other work depending on the desired request).

对了，还有个东西忘了讲了，我们有了task，构造了flow，那我们怎么去执行呢？这个时候就需要用到engine方法了，作为启动flow的引擎是绝对不能丢的！

engine里面自带一个run方法，run方法需要穿两个参数，第一个参数是flow，第二个参数就是flow里面的task所需要的参数，默认是以字典的形式传入的。

我们可以来看一个简单的例子：

import taskflow.engines
from taskflow.patterns import linear_flow as lf
from taskflow import task

class TaskA(task.Task):
def execute(self,str1,*args,**kwargs):
print('Hello,I am {0}'.format(str1))


class TaskB(task.Task):
def execute(self,str2,*args,**kwargs):
print('Hello,I am {0} years old'.format(str2))


if __name__ == "__main__":

 flow = lf.Flow('simple-linear').add(TaskA(), TaskB())
 taskflow.engines.run(flow,store = dict(str1='john', str2="23"))

最后附上一张官网的工作图：

over！

如果想了解更多taskflow的知识

你可以点击阅读原文

还可以扫一扫下面这个二维码