Flow
查看 MarkdownFlow 是 Dex 中最顶层的工作单元。一个 Flow type 描述一类业务流程，比如订单、订阅或审批。一个 Flow run 是它的一个实例，比如 42 号订单。
Dex 负责 Flow 的生命周期和状态。你的代码不需要数据库或任务队列，就能让流程在崩溃和重启之后继续运行。
什么时候用 Flow
每个需要由 Dex 管理流程的业务身份，用一个 Flow。共享这个身份和生命周期的工作，放进同一个 Flow 的 Step。一个 Flow 里的 Step 可以并行运行。
工作有自己的 owner 和生命周期时，启动另一个 Flow。两个 Flow 之间通过调用对方的 RPC 协作。父 Step 要等待子 Flow 时，用 SubFlow，例如把复杂逻辑拆成更小的 Flow，或者 fan out 工作。
保证
- 一个 Flow ID 同一时间最多只有一个 active run。
- 一个 Step 方法写入和发布的内容会一次全部生效，或者都不生效。见 Step 的保证。
- 使用默认的 SYNC durability 时，一个 Step 的结果在下一个 Step 开始前已经 durable 存储。Worker 崩溃和 Dex Server 重启都不会丢失它。
- 没有空闲的 Worker 时，Step 在 durable 队列里等待，有 Worker 可用时再运行。
Flow 包含什么
在你的语言里实现 Flow interface，就定义了一个 Flow。它声明三样东西：
- Steps：start Step，以及 Flow 能到达的其他所有 Step。Step 执行后台工作。
- Persistence schema：保存 Flow durable 状态的 Attribute 和 Channel。读写未声明的名称会被 Dex 拒绝。
- RPC handlers：client 在运行中的 Flow 上调用的方法。见 RPC。
start Step 的 input type 也是 Flow 的 start input type。下面这个 Flow 有两个 Step、一个 Attribute、一个 Channel 和一个 RPC。
status = Attribute("status", str)
notify = Channel("notify", None)
class ExampleFlow(Flow[int]):
def __init__(self) -> None:
self.finish = Finish()
self.example = Example(self.finish)
def get_steps(self) -> StepList[int]:
return StepList.start_step(self.example).other_steps(self.finish)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(status, notify)
@rpc
def describe(self, context: Context) -> RPCResult[str]:
return RPCResult.of(status.get(context))
例子: examples/python/dex_examples/primitives/flow/example_flow.py
标识符
"Flow" 这个词可以指 Flow 定义，也可以指一次 Flow run。下面这些标识符把它们区分开。
| 标识符 | 谁设置 | 含义 |
|---|---|---|
| FlowType | 应用（Flow 实现） | Dex 路由 Worker 调用和搜索结果使用的稳定类型名。 |
| Flow ID | 应用在 start 时 | 一次 Flow run 的业务身份。ID reuse policy 控制能否用同一个 Flow ID 再启动一个 run。 |
| Run ID | Dex Server | 这个 Flow ID 的一次 run 的身份。Time travel 保留它，并创建一个新的 run attempt。请用 Flow ID 定位 Flow；Run ID 只用于诊断。 |
| StepType | 应用（Step 实现） | Flow 内一个 Step 定义的稳定名称。 |
| StepExecutionID | Dex Server | 某个 Step type 的一次执行，格式为 StepType-Number。 |
大多数 SDK 默认用语言里的类型名作为 FlowType，你也可以显式覆盖。还有 Flow run 在运行时，不要改这些名称。
生命周期
Client 用一个 Flow ID 和 start input 启动 Flow。Dex 随后在 Worker 上运行 start Step。每个 Step 返回一个决定：进入下一个或多个 Step，或者关闭 Flow。同一时间可以有多个 Step 处于 active。
Flow 在以下情况下关闭：
- 某个 Step 完成 Flow，或让它失败；
- 它的 timeout 让它失败或取消它；
- client 停止它；
- 用同一个 Flow ID 的新启动终止它。只有新启动使用的 ID reuse policy 会终止正在运行的 Flow 时，才会这样。用 ALWAYS_RESTART 启动的 SubFlow 也会这样终止之前的子 Flow。
timeout 也可以不让 Flow 失败或取消，而是运行一个 handler，由它决定结果。Flow 运行期间，client 通过它的 RPC 读取和修改它。
Start options 和配置
两组选项决定一次 Flow run 的行为：
- StartFlowOptions 只在启动时生效一次。它设置 Flow timeout、ID reuse policy、start request ID 和 Attribute 初始值。
- FlowConfig 保存运行时设置，比如默认的 Step durability。Flow 运行期间可以更新。
用稳定的业务 Flow ID 直接启动 Flow。不要先读取或搜索，检查它是否已经存在。要安全地重试同一次启动，做三件事：使用拒绝第二个 run 的 ID reuse policy；每次重试都发送同一个 request ID；打开“返回已有 run，而不是报 already-started 错误”的启动选项。这样，重试一个已经成功的启动时，会拿到那个 run。如果 policy 允许再启动一个 run，重试就可能启动第二个 run。
参考
英文的 SDK 参考包含：