--- title: "Flow" description: "Flow 是 Dex 中最顶层的工作单元。一个 Flow type 描述一类业务流程，比如订单、订阅或审批。一个 Flow run 是它的一个实例，比如 42 号订单。" canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/flow/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/flow/index.mdx" locale: "zh-Hans" --- # Flow **Flow** 是 Dex 中最顶层的工作单元。一个 Flow type 描述一类业务流程，比如订单、订阅或审批。一个 Flow run 是它的一个实例，比如 42 号订单。 Dex 负责 Flow 的生命周期和状态。你的代码不需要数据库或任务队列，就能让流程在崩溃和重启之后继续运行。 ## 什么时候用 Flow 每个需要由 Dex 管理流程的业务身份，用一个 Flow。共享这个身份和生命周期的工作，放进同一个 Flow 的 [Step](https://docs.superdurable.io/zh-Hans/concepts/step.md)。一个 Flow 里的 Step 可以并行运行。 工作有自己的 owner 和生命周期时，启动另一个 Flow。两个 Flow 之间通过调用对方的 RPC 协作。父 Step 要等待子 Flow 时，用 [SubFlow](https://docs.superdurable.io/zh-Hans/concepts/primitives/subflow.md)，例如把复杂逻辑拆成更小的 Flow，或者 fan out 工作。 ## 保证 - 一个 Flow ID 同一时间最多只有一个 active run。 - 一个 Step 方法写入和发布的内容会一次全部生效，或者都不生效。见 [Step 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)。 - 使用默认的 SYNC durability 时，一个 Step 的结果在下一个 Step 开始前已经 durable 存储。Worker 崩溃和 Dex Server 重启都不会丢失它。 - 没有空闲的 Worker 时，Step 在 durable 队列里等待，有 Worker 可用时再运行。 ## Flow 包含什么 在你的语言里实现 **Flow** interface，就定义了一个 Flow。它声明三样东西： - **Steps**：start Step，以及 Flow 能到达的其他所有 Step。[Step](https://docs.superdurable.io/zh-Hans/concepts/step.md) 执行后台工作。 - **Persistence schema**：保存 Flow durable 状态的 [Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md) 和 [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md)。读写未声明的名称会被 Dex 拒绝。 - **RPC handlers**：client 在运行中的 Flow 上调用的方法。见 [RPC](https://docs.superdurable.io/zh-Hans/concepts/primitives/rpc.md)。 start Step 的 input type 也是 Flow 的 start input type。下面这个 Flow 有两个 Step、一个 Attribute、一个 Channel 和一个 RPC。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/flow/example_flow.py) **Go** ```go type ExampleFlow struct { dex.FlowDefaults } func (*ExampleFlow) GetSteps() []dex.StepDef { return []dex.StepDef{ dex.DefineStartStep(Example{}), dex.DefineStep(Finish{}), } } func (flow *ExampleFlow) GetRPCs() []dex.RPCDef { return []dex.RPCDef{dex.DefineRPC(flow.Describe, nil)} } func (*ExampleFlow) GetPersistenceSchema() dex.PersistenceSchema { return dex.PersistenceSchema{ Attributes: []dex.AttributeDef{Status}, Channels: []dex.ChannelDef{Notify}, } } func (*ExampleFlow) Describe(ctx dex.Context, _ dex.None) (*dex.RPCResult[string], error) { value, err := Status.Get(ctx) if err != nil { return nil, err } return &dex.RPCResult[string]{Output: value}, nil } ``` 例子: [examples/go/primitives/flow/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/flow/workflow.go) **Java** ```java public class ExampleFlow implements Flow { public static final Attribute status = Attribute.define("status", String.class); public static final Channel notify = Channel.define("notify", Void.class); @Override public StepList getSteps() { return StepList.startStep(exampleStep).otherSteps(finishStep); } @Override public PersistenceSchema getPersistenceSchema() { return PersistenceSchema.of(status, notify); } @RPC public RPCResult describe(final Context context) { return RPCResult.of(status.get(context)); } } ``` 例子: [examples/java/src/main/java/io/superdurable/dex/primitives/flow/ExampleFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/flow/ExampleFlow.java) **TypeScript** ```typescript export class ExampleFlow implements Flow { public getSteps() { return StepList.startStep(this.example).otherSteps(this.finish); } public getPersistenceSchema(): PersistenceSchema { return { attributes: [status], channels: [notify] }; } @rpc({ outputCodec: stringCodec }) public describe(context: Context): RPCResult { return { output: status.get(context) ?? "" }; } } ``` 例子: [examples/typescript/src/primitives/flow/example-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/flow/example-flow.ts) **Rust** ```rust impl Flow for ExampleFlow { type StartInput = i32; fn steps(&self) -> StepList<'_, Self::StartInput> { StepList::start(&self.example).and(&self.finish) } fn persistence(&self) -> PersistenceSchema { PersistenceSchema::new() .attribute(&status()) .channel(¬ify()) } fn rpcs(&self) -> RpcList { RpcList::new().function(DESCRIBE, Self::describe) } } ``` 例子: [examples/rust/src/primitives/flow/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/flow/flow.rs) ## 标识符 "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](https://docs.superdurable.io/zh-Hans/concepts/client.md) 用一个 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 参考包含： - [Flow](https://docs.superdurable.io/zh-Hans/reference/python/flow.md) 定义； - [StartFlowOptions](https://docs.superdurable.io/zh-Hans/reference/python/start-flow-options.md)，以及 [Flow timeout 和 timeout handler](https://docs.superdurable.io/zh-Hans/reference/python/start-flow-options.md#timeout-and-timeout-policy)； - [Flow ID、RequestID 和 already-started 处理](https://docs.superdurable.io/zh-Hans/reference/python/start-flow-options.md#flow-id-request-id-and-already-started)； - [StartFlow 最佳实践](https://docs.superdurable.io/zh-Hans/reference/python/start-flow-options.md#start-flow-best-practices)； - [FlowConfig](https://docs.superdurable.io/zh-Hans/reference/python/flow.md#flow-config) 以及如何更新它。