SubFlow

查看 Markdown查看 Markdown

SubFlow 是父 Flow 从它的某个 Step 启动的 Flow run。子 Flow 是一个普通的 Flow，有自己的 Step、Attribute 和 Channel。父 Flow 像等待 WaitFor 中的其他 condition 一样等待它。

什么时候用 SubFlow

SubFlow 主要有两种用途：

1. 把复杂逻辑拆到更小的 Flow 中。这让业务逻辑更容易维护。
2. 扇出工作。一个 Flow run 不能同时运行太多操作，但可以把工作交给很多 SubFlow。这些 SubFlow 同时运行，父 Flow 因此能达到高得多的并发。

保证

Dex Server 根据父 Flow ID、Step execution 和 condition 序号计算 SubFlow ID。因此重试或重复启动会找到同一个子 Flow，而不是再启动一个。默认情况下，父 Flow 继续等待运行中的子 Flow，使用已完成子 Flow 的结果，并重启异常结束的子 Flow。

SubFlow ID 用短横线连接各个部分。所以 Step type 含短横线的 Step 不能启动 SubFlow，否则两个父 Flow 可能算出同一个 SubFlow ID，共用一个子 Flow。Dex 会让这样的 WaitFor attempt 直接失败，不再重试。启动 SubFlow 的 Step，Step type 里不要有短横线。

用同一个父 Flow ID 再启动一次时，新的 run 通常会算出和之前的 run 相同的 SubFlow ID。这时 Dex 按 reuse policy 处理。使用默认 policy 时，新的 run 会拿到之前 run 的子 Flow（无论它还在运行还是已经完成），直到 Dex 删除那个子 Flow。要让每个 run 都启动新的子 Flow，就用 ALWAYS_RESTART。如果之前的子 Flow 还在运行，它会被终止。两个子 Flow 都要继续运行时，换一个父 Flow ID。

父 Step 被取消或父 Flow 关闭时，子 Flow 会继续运行。你需要自己停止它，或者给子 Flow 设置 Flow timeout，让它到时自行结束。

启动并等待 SubFlow

目标 Flow 必须注册在 Worker 上。它的 start Step 必须接受传给 SubFlow 的输入。WaitFor 启动子 Flow。子 Flow 完成后，Execute 读取它的结果。

parent_flow.pyPython

123456789101112131415

class SubFlowParent(Step[int]):
    def __init__(self, target: Flow[int]) -> None:
        self.target = target
        self.options = SubFlowOptions(
            timeout=timedelta(hours=1),
            timeout_policy=FlowTimeoutPolicy.CANCEL,
        )

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.until(SubFlow.run(self.target, input, self.options))

    def execute(self, context: Context, input: int) -> StepDecision:
        result = SubFlow.get_condition_results(context)
        output = result.single_output(int)
        return graceful_complete(f"{SubFlow.get_flow_id(context)}|{output}")

例子: examples/python/dex_examples/primitives/subflow/parent_flow.py

等待多个 SubFlow

用 allOf 让父 Flow 等待所有 SubFlow 完成。

父 Flow 不需要等待全部 SubFlow 时，用 anyOf。WaitFor 停止等待后，SubFlow 仍会继续运行。Execute 仍然可以拿到 SubFlow ID 来停止它，或者把 ID 传给另一个 Step 决定如何处理。SubFlow 与其他 condition 的组合见 SubFlow results。

参考

每个 SDK 参考的 SubFlow 页列出该语言的 API 名称。

SubFlowOptions 接受与 Client 启动 Flow 相同的选项，另加一个 reuse policy。选项和面对已有子 Flow 各种状态时的 reuse policy，见 SubFlowOptions 和 SubFlowReusePolicy。