前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Taskflow 有兴趣了解一下?

Taskflow 有兴趣了解一下?

作者头像
我被狗咬了
发布2019-09-23 11:21:57
4.7K0
发布2019-09-23 11:21:57
举报
文章被收录于专栏:Python乱炖

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

声明:

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

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

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

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

具体来说就是:

代码语言:javascript
复制
from taskflow import task
代码语言:javascript
复制
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)

代码语言:javascript
复制
from taskflow.patterns import linear_flow
代码语言:javascript
复制
linear_flow.Flow('linear').add(
    taskA(),
    taskB(),
    taskC()
)

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

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

回答:当然是!

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

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

代码语言:javascript
复制
from taskflow.patterns import unordered_flow
代码语言:javascript
复制
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之间存在依赖关系,那我们这个时候就可以使用图流了:

实现:

代码语言:javascript
复制
from taskflow.patterns import graph_flow
代码语言:javascript
复制
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.

代码语言:javascript
复制
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所需要的参数,默认是以字典的形式传入的。

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

代码语言:javascript
复制
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!

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2019-01-09,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Python乱炖 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • Conceptual example
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档