跳到主要内容
Super Durable
简体中文
English简体中文
GitHub登录注册 Cloud
菜单

Step

查看 Markdown

Step 是 Flow 里的一个后台工作单元。它可以先 durable 地等待某件事发生，再执行工作，然后决定下一步运行什么。Dex 在你的 Worker 上运行 Step，并记录每个结果。

Step 由一个原生 struct 或 class 实现，它的类型就是 StepType。同一个 Step type 在一个 Flow 里可以运行多次，甚至并行运行。每次运行是一个 StepExecution，用 StepExecutionID 标识，比如 Charge-2。

Step input​

每个 Step execution 有自己的 typed input，Dex 把它和 execution 一起保存。input 用来区分同一个 Step type 的不同 execution。start Step 的 input 就是 Flow 的 start input。

WaitFor 和 Execute​

一个 Step 有一个或两个方法。

  • WaitFor 可选。它返回一个 durable 的 Wait。Dex 保存这个 wait，替你检查条件，条件满足后调用 Execute。
  • Execute 必需。它执行工作，并返回一个 StepDecision。

Step 没有 WaitFor 时，Execute 立即运行。

下面这个 Step 等待一条审批消息，然后进入第二个 Step，由它完成 Flow。

step_flow.pyPython
approval = Channel("Approval", str)


class StepSecond(Step[int]):
    def execute(self, context: Context, input: int) -> StepDecision:
        return graceful_complete(input + 1)


class Example(Step[int]):
    def __init__(self, second: StepSecond) -> None:
        self.second = second

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.until(approval.for_one())

    def execute(self, context: Context, input: int) -> StepDecision:
        return go_to(StepSecond, input + 1)

例子: examples/python/dex_examples/primitives/step/step_flow.py

等待​

等待不占用 Worker 或线程。Dex 把它保存在 Flow 的状态里，条件满足时唤醒 Step。Step 可以等待三类条件：

用 AnyOf 或 AllOf 组合条件。AnyCombinationOf 在几组条件中任意一组全部满足时成立。等待结束后，Execute 可以读取满足它的结果，比如它消费的 Channel 消息。见 Waits。

要等待并行分支，用 Join。它不是另一类条件：它等待的是各分支上报结果的 ChannelMap。见 等待并行 Step。

StepDecision​

Execute 返回一个 StepDecision。它可以进入下一个 Step，也可以同时进入多个 Step。分支、循环和并行工作都这样构建。

决定也可以关闭 Flow 或结束一个分支：

  • GracefulComplete 请求 Dex 在没有其他 active Step 时完成 Flow。
  • ForceComplete 立即完成 Flow。
  • ForceFail 立即让 Flow 失败。
  • DeadEnd 结束这个分支，但不关闭 Flow。

完成类的决定可以返回 Flow output，client 把它作为 Flow result 读取。多个完成类的决定都返回 output 时，result 会保留每一个。只有最终完成的 Flow 才保留这些 output。如果 Flow 没有完成（失败、被取消、被终止或超时），它的 result 不含任何 output。

决定还可以取消其他 active Step。见 Parallel Steps 和 Parallel SubFlows 模式。

StepOptions​

Step 返回 StepOptions，控制它的方法如何运行。WaitFor 和 Execute 各有自己的设置。

  • Timeout 限制方法的一次 attempt。
  • Retry policy 在 attempt 失败后再次运行同一个方法。
  • Failure policy 决定重试用完之后怎么办。默认让 Flow 失败。Execute 可以改为进入一个 recovery Step，WaitFor 可以继续执行 Execute。见 WaitFor failure policy、Execute failure policy 和 Last failure info。
  • Durability 决定 Dex 什么时候存储结果。默认的 SYNC 在下一步工作开始前存储。ASYNC 更快，但 server 故障可能让方法再运行一次。
  • Heartbeat 让长时间运行的方法报告进度并保存 checkpoint。
  • State loading 和 Attribute locks 决定方法加载哪些状态、锁住哪些 Attribute。

StepDecision 可以为某一次 Step execution 覆盖这些选项。

保证​

Dex 单独保存每次 WaitFor 或 Execute 调用。它把这次调用的 Attribute 写入和 Channel 发布与结果一起保存，而且只在方法成功时保存。对外部系统的调用不在这次保存里。

一个 Step 方法的修改会一次全部生效。但这挡不住其他 Step 和 RPC 同时修改同样的状态。方法读到的是 Dex 把它交给 Worker 时的 Flow 状态。在方法的修改生效之前，其他 Step 和 RPC 可能已经改了这些状态。Dex 不检查方法读到的值是否仍然是最新的。所以两个先读后写同一个 Attribute 的 Step 或 RPC，可能互相覆盖对方的更新。要避免这种情况，让它们都在这个 Attribute 上声明 lock。Dex 会让它们依次执行。见 Attribute lock。

一个 Step 方法可能运行不止一次：失败后的重试会再运行它。Dex 不一定知道失败的 attempt 有没有执行工作。请让每个外部副作用都幂等，用 Flow ID 加 StepExecutionID，或你自己的业务 key 作为幂等 key。Dex 无法判断的情况见 Failure and redelivery。

重试用完时，recovery Step 能知道工作是肯定没有运行，还是可能运行过。

取消一个 Step，会丢弃它当前方法还没有保存的一切：Attribute 写入、Channel 发布和它的决定。它的 WaitFor 已经保存的内容保留。一个决定或 RPC 永远不会取消它自己启动的 Step。被取消的 Step 已经消费的消息会怎样，见 Channel 的保证。

Dex 不会强行停止被取消方法的代码。方法应检查取消状态并返回。在它返回之前，它的代码会继续运行。它对外部系统的调用仍然会生效，即使 Flow 已经关闭。

GracefulComplete 结束它自己的分支，并请求 Dex 完成 Flow。一旦没有 active Step，Dex 就开始完成 Flow。在此之前，其他 Step 和 RPC 仍然可以启动新的 Step，这些 Step 也会推迟完成。之后某个分支以 DeadEnd 结束，也不会撤回这个请求。最后一个 Step 结束后，Dex 还要处理完待写的 Attribute Sync 和已经开始的 transactional RPC，所以 Flow 可能会短暂保持 RUNNING。

没有待处理的 GracefulComplete 时，一个没有 active Step 的 Flow 会继续运行。RPC 仍然可以在里面启动新的 Step。用一个完成类的决定结束最后一个分支。如果 Flow 不能无限等待，就给它设置 Flow timeout。

Dex 大致按 Step 就绪的顺序启动它们，但不严格保证。不要依赖并行 Step 启动或结束的顺序。顺序重要时，用 Channel 或 Attribute lock。

参考​

英文的 SDK 参考包含：