Python工作流设计模式

Question

我正在设计一个软件，但现在有点迷茫，不知道自己在做什么，同时又觉得自己在重复别人已经做过的事情。

我的情况是这样的：我正在设计一个科学工具，里面有一个互动的用户界面。用户的输入应该能引发视觉反馈（这很明显），有些反馈是直接的，比如编辑一个区域的形状，有些则是尽快给出反馈，而不影响用户的操作，比如在这个区域内解决一些偏微分方程。

如果我把所有需要执行的操作画成一个图，我会得到一个非常复杂的图，里面展示了很多可以并行处理和缓存/重用部分结果的机会。所以我主要想利用这种并行处理的方式（让选定的子任务在不同的进程中执行，结果自动被后续任务合并，后续任务在等待所有输入准备好时），而且只需要重新计算那些输入发生变化的部分。

pyutilib.workflow似乎是最接近我想要的东西，当然它并没有（看起来没有进行任何子进程处理）。这让我有点失望；虽然我不是软件工程师，但我觉得我并没有提出什么过分的要求。

另一个复杂的因素是我希望有紧密的用户界面集成，而其他科学工作流解决方案似乎并不支持这一点。例如，我想通过一个转换节点传递一个拖放事件以进行进一步处理。这个转换节点有两个输入；一个是仿射变换状态输入端口，另一个是知道如何处理它的点集类。如果仿射变换输入端口是“脏的”（等待其依赖项更新），事件应该被暂时搁置，直到它变得可用。但是当事件通过节点后，事件输入端口应该标记为已处理，这样当仿射变换因用户进一步输入而改变时，就不会再次触发。这只是我看到的许多问题中的一个例子，我没有看到任何地方对此进行讨论。或者当一个长时间运行的分叉-合并分支在处理之前的输入时接收到新输入，该怎么办。

所以我的问题是：你知道有哪些好的书籍或文章可以阅读关于工作流设计模式吗？还是说我在试图把一个方钉放进圆孔，而你知道一种完全不同的设计模式我应该了解？或者有没有一个Python包可以实现我想要的功能，不管它的宣传词是什么？

我在enthought.traits的基础上自己做了一个解决方案，但我对此也不是特别满意，因为感觉像是在粗糙地重复别人的工作。只是我在网上找不到任何现成的解决方案。

注意：我并不想要网页框架、图形工作流设计工具或任何特殊用途的工具。只想要一些概念上类似于pyutilib.workflow的东西，但包括文档和我可以使用的功能。

# # # 编辑：这是我在更多阅读和思考后得出的结论：

可以加在“工作流架构”上的需求太多样化，根本没有一种解决方案能适合所有情况。你想要与磁盘存储紧密集成，还是与网页框架紧密集成，或者需要异步处理，甚至混合自定义有限状态机逻辑来调度任务？这些都是有效的需求，但它们大多不兼容，或者会导致毫无意义的混合。

不过，并不是一切都失去了希望。寻找一个通用的工作流系统来解决任意问题，就像寻找一个通用的迭代器来解决你的自定义迭代问题。迭代器并不是主要为了可重用性；你不能用红黑树的迭代器来遍历你的张量。它们的强项在于关注点的清晰分离和统一接口的定义。

我想要的（并且已经开始自己编写；这将会很酷）看起来是这样的：它的基础是一个与实现无关的工作流声明迷你语言，基于装饰器和一些元魔法，将下面这样的语句转换成包含所有所需信息的工作流声明：

@composite_task(inputs(x=Int), outputs(z=Float))
class mycompositetask:
    @task(inputs(x=Int), outputs(y=Float))
    def mytask1(x):
        return outputs( y = x*2 )
    @task(inputs(x=Int, y=Float), outputs(z=Float))
    def mytask2(x, y):
        return outputs( z = x+y )
    mytask1.y = mytask2.y   #redundant, but for illustration; inputs/outputs matching in name and metadata autoconnect

装饰器返回的是一个任务/复合任务/工作流的声明类。除了类型约束外，其他所需的元数据也可以轻松添加到语法中。

现在这个简洁且符合Python风格的声明可以传递给一个工作流实例工厂，返回实际的工作流实例。这个声明语言相当通用，可能在不同的设计需求之间不需要太多变化，但这样的工作流实例化工厂完全取决于你的设计需求和想象力，除了提供/获取输入/输出的公共接口。

在最简单的情况下，我们会有这样的东西：

wf   = workflow_factory(mycompositetask)
wf.z = lambda result: print result   #register callback on z-output socket
wf.x = 1    #feed data into x input-socket

其中wf是一个简单的工作流实例，它只是在同一线程上将所有包含的函数体串联在一起，一旦所有输入都绑定。这是一个相当冗长的方式来串联两个函数，但它说明了这个想法，并且已经实现了将信息流的定义集中在一个地方，而不是分散在那些不想与之有任何关系的类中。

这大致就是我目前实现的功能，但这意味着我可以继续我的项目，并且在适当的时候我会添加对更复杂工作流实例工厂的支持。例如，我在考虑分析依赖图，以识别分叉和合并，并跟踪每个输入在工作流实例级别上产生的活动，以实现优雅的负载均衡和取消那些失去相关性的特定输入的影响，但仍然占用资源。

无论如何，我认为将工作流声明、接口定义和实例化实现分开的项目是值得努力的。一旦我有几个非平凡的工作流实例类型运行良好（我意识到我至少需要两个来完成我正在进行的项目*），我希望能找到时间将其作为一个公共项目发布，因为尽管工作流系统的设计需求多种多样，但有了这个基础，实施你自己的特定需求会简单得多。而且，与其有一个臃肿的工作流框架，不如围绕这样的核心发展出一个可以轻松替换的自定义解决方案的瑞士军刀。

*意识到我需要将代码拆分为两种不同的工作流实例类型，而不是试图将所有设计需求压缩到一个解决方案中，这让我心中的方钉和圆孔变成了两个完美互补的孔和钉。

Answer 1

我觉得你们的疑虑是对的，也有些不对，关于是否要重新发明轮子。也许不同的思维层次能给你一些启示。

怎么吃掉一头大象呢？

层次A：软件设计

在这个层次上，你需要遵循一个最佳实践，那就是不要在用户界面（UI）上进行长时间的操作（也就是不要在UI线程上做这些事情）。你需要一个专注于收集输入（包括取消操作）和绘制（包括进度条或沙漏等正在进行的可视化）的UI层。这个层次应该和其他任何东西分开得像黑夜和白天一样。任何在这个层次之外的调用都必须快速，这样才能让用户感觉直观和响应迅速。

在像你这样复杂的任务中，UI层之外的调用通常包括：

1. 安排一些工作——命令应该被排队到智能层，以便它在合适的时候处理。
2. 读取结果——结果应该在智能层中排队，这样它们就可以被“弹出”并呈现。
3. 取消/停止/退出——只需设置一个标志。智能层应该不时检查这个标志。

不要太担心某些用户操作反应得太慢——如果你有一个坚实的设计核心，你可以在后面调整用户输入的优先级。或者添加一个短期的沙漏或类似的东西。甚至可以取消所有在特定用户输入后变得过时的长时间操作。

层次B：重负担的智能层

没有“最佳”的框架适用于“任何”类型的重工作。

所以我建议你把这个层次的输入（通过UI）设计得尽可能简单，不要涉及框架。

在内部，你可以使用一些框架来实现，但将来你会有能力根据需要重新设计这些重负担的元素。例如，将来你可以：

• 把一些数学运算交给GPU
• 把任务分配给服务器集群
• 利用云计算

对于复杂的任务，在设计的顶层选择一个框架可能会在未来成为障碍。具体来说，它可能会限制你使用其他技术的自由。

很难确定，但我觉得你没有一个“灵丹妙药”的框架来解决你的任务。所以你应该找到强大的工具（例如线程和队列）来实施良好的设计实践（例如解耦）。

编辑：作为对你编辑的回应

你最新的编辑完美地强调了软件设计师面临的艰巨挑战。对于你的情况，接受没有灵丹妙药的事实。我建议你尽早接受这个事实——越早越好……

提示在于你提供了最通用的任务，由整数和浮点数来定义。这可能让你今天感到高兴，但明天就会失败。就像锁定在一个超级抽象的框架中一样。

这个方向是对的——在你的设计中有一个重负担的“任务”基础。但它不应该定义整数或浮点数。相反，专注于上面提到的“开始”、“读取”和“停止”。如果你看不到你正在吃的大象的大小，那么你可能会失败，最终饿肚子 :)

从层次A的设计角度来看，你可以定义任务包含如下内容：

class AnySuperPowerfulTask:
    def run():
        scheduleForAThreadToRunMe()
    def cancel():
        doTheSmartCancellationSoNobodyWouldCrash()

这为你提供了基础——中立、干净，并且从层次A（设计）的角度来看是解耦的。

不过，你需要某种方式来设置任务并获取真实结果，对吧？当然，这将属于层次B的思考。这将是特定于某个任务（或作为中间基础实现的一组任务）。最终任务可能是这样的：

class CalculatePossibilitiesToSaveAllThePandas(SuperPowerfulTask):
    def __init__(someInt, someFloat, anotherURL, configPath):
        anythingSpecificToThisKindOfTasks()
    def getResults():
        return partiallyCalculated4DimensionalEvolutionGraphOfPandasInOptimisticEnvoronment()

（这些示例故意在Python中不正确，以便专注于设计，而不是语法）。

层次C - 抽象的极乐世界

看起来这个层次应该在这篇文章中提到。

是的，许多优秀设计师都能确认，这里有一个陷阱。你可能会无休止地（而且没有任何结果）寻找“通用解决方案”，也就是灵丹妙药。我建议你看看这个，然后快点离开，别让自己陷得太深 ;) 陷入这个陷阱并不可耻——这是伟大设计师的正常发展阶段。至少我试着相信这一点 :)

编辑2

你说：“我正在进行一项软件设计，我卡住了，不知道自己在做什么，感觉像是在重新发明轮子。”

任何软件设计师都有可能卡住。也许下一个思考层次能帮到你。接下来是：

层次D - 我卡住了

建议：离开这个地方。走到街角的咖啡馆，点一杯最好的咖啡，坐下来。问自己“我需要什么？”注意，这与“我想要什么？”是不同的。不断问这个问题，直到你排除掉错误的答案，并开始观察正确的答案：

错误的答案：

1. 我需要一个可以做X、Y和Z的框架。
2. 我需要一个可以以200英里每小时运行并在附近的农场收割森林的螺丝刀。
3. 我需要一个用户实际上永远看不见的惊人内部结构。

正确的答案（如果我理解你的问题有误，请原谅我）：

1. 我需要用户能够给软件输入。
2. 我需要用户看到计算正在进行。
3. 我需要用户可视化计算结果。