Taskflow 有兴趣了解一下?

网友投稿 904 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()

Taskflow 有兴趣了解一下?

)

最后还有一种流是图流(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小时内删除侵权内容。

上一篇:Web Service进阶(五)SOAPBinding方式讲解
下一篇:基于FusionInsight开发智能搜车系统
相关文章