Documentation
¶
Overview ¶
Package workflow 提供组合式流程编排框架。
支持串行、并行、异步、条件分支四种流程模式，可任意嵌套组合。 所有流程节点共享 Runnable 接口，Hook 自动递归传播实现 AOP。
流程类型 ¶
- PipeFlow: 串行流程，按顺序依次执行，任一失败则停止
- ParallelFlow: 并行流程，并发执行，任一失败则取消其他
- AsyncFlow: 异步流程，返回 Future，不阻塞调用方
- SwitchFlow: 条件分支，根据 key 选择执行路径
组合嵌套 ¶
pipe := workflow.NewPipe() pipe.Add(step1, step2) parallel := workflow.NewParallel(10) parallel.Add(task1, task2) // 嵌套：串行流程中包含并行流程 pipe.Add(parallel)
Context 取消传播 ¶
所有流程支持 context 取消传播：
ctx, cancel := context.WithCancel(context.Background()) defer cancel() // 如果 ctx 被取消，所有正在执行的任务会收到取消信号 err := pipe.Run(ctx)
任务错误与 context 取消的语义 ¶
ParallelFlow/AsyncFlow 对错误链包含 context.Canceled/DeadlineExceeded 的 任务错误采用统一豁免规则：
- 流程级 context 确实被取消/超时 → 视为取消的副作用，不上报；
- 流程级 context 健康 → 视为任务自身的失败（例如任务内部子操作超时 返回 fmt.Errorf("db timeout: %w", context.DeadlineExceeded)），必须 正常上报给调用方，绝不吞错。
Hook AOP ¶
hook := workflow.NewHookManager()
hook.Add(&LogHook{}, &MetricsHook{})
// Hook 会自动递归传播到子流程
pipe.SetHookManager(hook)
异步任务 ¶
async := workflow.NewAsync(10) async.Add(task1, task2, task3) // 返回 Future，不阻塞 future := async.RunAsync(ctx) // 等待完成或取消 err := future.Wait(ctx) future.Cancel()
Index ¶
- type AsyncFlow
- func (a *AsyncFlow) Add(tasks ...Runnable) *AsyncFlow
- func (a *AsyncFlow) Hooks() *HookManager
- func (a *AsyncFlow) Run(ctx context.Context) error
- func (a *AsyncFlow) RunAsync(ctx context.Context) *Future
- func (a *AsyncFlow) RunAsyncWithResults(ctx context.Context) <-chan AsyncResult
- func (a *AsyncFlow) SetHookManager(hm *HookManager)
- type AsyncResult
- type Flow
- type Future
- type Hook
- type HookManager
- type ParallelFlow
- type PipeFlow
- type Runnable
- type SwitchFlow
- type SwitchFunc
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AsyncFlow ¶
type AsyncFlow struct {
// contains filtered or unexported fields
}
AsyncFlow 异步执行多个任务。 返回 Future，不阻塞调用方。 支持 context 取消传播和任务取消。
func (*AsyncFlow) Add ¶
Add 添加一个或多个异步任务。 若任务本身是子流程,当前 HookManager 会同步传播给它(与 PipeFlow.Add 对称,2026-09-06 复审修复)。
func (*AsyncFlow) RunAsync ¶
RunAsync 异步执行所有任务，返回 Future。 调用方可以通过 Future.Wait() 等待完成，或 Future.Cancel() 取消任务。
func (*AsyncFlow) RunAsyncWithResults ¶
func (a *AsyncFlow) RunAsyncWithResults(ctx context.Context) <-chan AsyncResult
RunAsyncWithResults 异步执行所有任务，返回结果通道。 每个任务完成后会发送结果到通道。
func (*AsyncFlow) SetHookManager ¶
func (a *AsyncFlow) SetHookManager(hm *HookManager)
SetHookManager 设置 HookManager（用于嵌套传播）。
type Flow ¶
type Flow interface {
Runnable
Hooks() *HookManager
SetHookManager(*HookManager)
}
Flow 流程容器接口。 继承 Runnable，支持组合嵌套。 SetHookManager 用于递归传播 HookManager 到子流程。 Hooks 返回当前 manager(传播前据此检测子流程自有 hook 被覆盖并告警)。
type Future ¶
type Future struct {
// contains filtered or unexported fields
}
Future 异步任务的结果。 支持等待完成、检查状态、取消。
type Hook ¶
type Hook interface {
// Before 在 Runnable 执行前调用。
// 返回 error 时中止执行（HookManager.Before 短路传播）。
Before(ctx context.Context, r Runnable) error
// After 在 Runnable 执行后调用（包括 Run panic 时通过 defer 触发）。
// err 为 nil 表示执行成功，非 nil 表示执行失败。
After(ctx context.Context, r Runnable, err error)
}
Hook 生命周期钩子接口。 用于在 Runnable 执行前后插入通用逻辑（如日志、监控、链路追踪、权限校验）。
type HookManager ¶
type HookManager struct {
// contains filtered or unexported fields
}
HookManager 钩子管理器。 按注册顺序依次执行所有 Hook，支持递归传播到子流程。
并发契约：Add 非 goroutine 安全——所有 Hook 必须在 Run 之前注册完毕 （Run 期间 Add 是数据竞争）。Run 期间同一 Hook 会被并发调用 （Parallel/Async 多 goroutine），Hook 实现必须并发安全且自担幂等； 嵌套流程会逐层各触发一次（非端到端一次）。
func (*HookManager) After ¶
func (m *HookManager) After(ctx context.Context, r Runnable, err error)
After 按注册顺序依次调用所有 Hook 的 After 方法。
type ParallelFlow ¶
type ParallelFlow struct {
// contains filtered or unexported fields
}
ParallelFlow 并发执行多个任务。 支持并发限制和 context 取消传播。 当任一任务失败时，取消其他正在执行的任务。
func NewParallel ¶
func NewParallel(maxProcs int) *ParallelFlow
NewParallel 创建一个并行流程。 maxProcs 为最大并发数，<=0 表示不限制。
func (*ParallelFlow) Add ¶
func (p *ParallelFlow) Add(tasks ...Runnable) *ParallelFlow
Add 添加一个或多个并发任务。 若任务本身是子流程,当前 HookManager 会同步传播给它(与 PipeFlow.Add 对称,2026-09-06 复审修复)。
func (*ParallelFlow) Run ¶
func (p *ParallelFlow) Run(ctx context.Context) error
Run 并发执行所有任务。 返回第一个非 nil error，或 context 取消错误。
错误豁免规则（与 AsyncFlow 一致）：仅当流程级 context 确实被取消/超时 时，错误链中的 context.Canceled/DeadlineExceeded 才视为取消的副作用被 豁免；流程 context 健康时，任务自身的超时/取消失败必须上报。
func (*ParallelFlow) SetHookManager ¶
func (p *ParallelFlow) SetHookManager(hm *HookManager)
SetHookManager 设置 HookManager（用于嵌套传播）。
type PipeFlow ¶
type PipeFlow struct {
// contains filtered or unexported fields
}
PipeFlow 顺序执行多个步骤。 任一步骤失败则停止并返回错误。 支持 context 取消传播。
func (*PipeFlow) Add ¶
Add 添加一个或多个步骤（按添加顺序执行）。 若步骤本身是子流程,当前 HookManager 会同步传播给它 — "先 SetHookManager 后 Add"与"先 Add 后 SetHookManager"两种顺序下 hook 均生效(2026-09-06 复审修复:原实现只在 SetHookManager 时传播)。
func (*PipeFlow) Run ¶
Run 顺序执行所有步骤。 任一步骤失败或 context 取消则停止并返回错误。 Hook 或 Runnable 的 panic 会被捕获并转为 error 返回。
func (*PipeFlow) SetHookManager ¶
func (p *PipeFlow) SetHookManager(hm *HookManager)
SetHookManager 设置 HookManager（用于嵌套传播）。
type Runnable ¶
type Runnable interface {
// Run 执行业务逻辑。
// 返回 error 表示执行失败，会触发流程中断或取消。
Run(ctx context.Context) error
}
Runnable 可执行单元的基础接口。 所有流程节点（叶子节点和容器 Flow）都必须实现此接口。
type SwitchFlow ¶
type SwitchFlow struct {
// contains filtered or unexported fields
}
SwitchFlow 条件分支流程。 根据 SwitchFunc 返回的 key 选择执行路径。
func (*SwitchFlow) Case ¶
func (s *SwitchFlow) Case(key string, runnable Runnable) *SwitchFlow
Case 添加一个分支。 重复 key 后者覆盖前者（静默）；分支集合应在 Run 前定型。 若分支本身是子流程,当前 HookManager 会同步传播给它(2026-09-06 复审修复)。
func (*SwitchFlow) Run ¶
func (s *SwitchFlow) Run(ctx context.Context) error
Run 根据 SwitchFunc 选择并执行分支。 如果 SwitchFunc 返回的 key 不存在，返回错误。 SwitchFunc 自身的 panic 同样被捕获转为 error（与 PipeFlow 的 "Hook 或 Runnable 的 panic 会被捕获"契约对齐,2026-09-06 复审修复: 原实现裸调 switchFunc,分支选择函数 panic 直接炸穿调用方）。
func (*SwitchFlow) SetHookManager ¶
func (s *SwitchFlow) SetHookManager(hm *HookManager)
SetHookManager 设置 HookManager（用于嵌套传播）。
type SwitchFunc ¶
SwitchFunc 条件路由函数。 根据 context 返回要执行的分支 key。
Source Files