React(一)react概述、组件、事件
921
2022-05-30
最近在工作中会经常使用到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的知识
你可以点击阅读原文
还可以扫一扫下面这个二维码
SQL
版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。