README
¶
workflow
组合式流程编排——四种流 + Hook AOP
🌐 项目主页：https://oekit.github.io/ · 名称由来：Open Easy Kit（开源 · 易用 · 工具集合） 🤖 AI Agent：本仓规范与硬规则见 AGENTS.md；机器可读索引见 llms.txt（接口签名以源码为准）。
安装
go get github.com/oekit/workflow@v0.3.1
这个仓库做什么
workflow 是组合式流程编排框架：串行/并行/异步/条件分支四种模式任意嵌套，
Hook AOP 自动向子流程递归传播，context 取消自动传递，panic 全量 recover。
四种流
| 流 | 语义 |
|---|---|
PipeFlow |
串行管道：步骤依次执行，任一失败即中止 |
ParallelFlow |
并行：并发执行（可限并发数），聚合错误 |
AsyncFlow |
异步：提交后收集结果，错误经 resultFn 上抛 |
SwitchFlow |
条件分支：按判定选择子流程 |
快速使用
import "github.com/oekit/workflow/workflow"
p := workflow.NewPipe()
p.Add(step1, step2, workflow.NewParallel(4))
err := p.Run(ctx)
Hook AOP
HookManager 注册 Before/After/成功/失败钩子，自动递归传播到所有子流程——
审计日志、链路打点、熔断注入都不用改业务步骤。
边界
- 步骤是任意
Runnable（Run(ctx) error）；定时触发用 timertask
矩阵总览（24 活跃模块，自动生成）
| 模块 | 内容 | 矩阵内依赖 |
|---|---|---|
| pkg | 微工具集 19 合 1（errors/bloomx/contextx/container/stringx/slicex/enumx/convert/encoding/jsonrepair/httpjson/mathx/pathx/pagination/ptrx/timex/cache/bufiox/strmap；v0.10.0 起 uidgen/validator 拆至 kitx）；cache（v0.4.0 起）含缓存接口 CacheAside + BatchGetter 批量能力（MGet）+ LocalCache/SingleFlightCache + cachetest 契约 + Checker；container（v0.9.0 起，收编原 collectionx）泛型容器 Set/Ring/Queue/SafeMap + Map（sync.Map 背书）/Slice（并发安全）；timex/stringx（v0.7.0 起）时间段边界 BeginningOf/EndOf 七对（分/时/日/周/月/季/年，周起始显式参数）+ trie 多模式替换器 Replacer（最长匹配）/ NewMask 敏感词遮罩；singleflighttest（v0.8.0 起）singleflight 共享契约测试套件（internal 与 concurrency 双侧引用） | — |
| concurrency | 并发与流控原语（async/gopool/group/singleflight/progress/syncx/ratelimit/retry）+ window 滑动窗口 / breaker 自适应熔断（规范哨兵 ErrCircuitOpen）/ shedding CPU 自适应降载 / batch 攒批执行器（满条数/满字节/窗口去重）/ timewheel 单级时间轮；v0.5.0 起 retry 指数退避+抖动（MaxDelay 封顶/溢出饱和）、loop 周期原语（忙时丢 tick/单轮 panic 收编/FirstDelay 可打断）、async labeled RunSafe（panic 按任务标签上报）、syncx.FanInErr 逐槽错误（panic 槽显式标记）、KeyedLimiter 增强（容量护栏/RetryAfter/速率热更）并接入 pkg/singleflighttest 契约测试 | pkg |
| logx | 统一日志门面（多后端可替换 + Sink + caller 归属）+ 优雅关停；桥接子包 hlog/hertzlog/kitexlog 为嵌套模块按需引入 | — |
| core | 应用基建件（governance 治理 schema / config 配置加载 + default/env/range/options tag 校验 / lifecycle 契约 / network IP 工具 / version） | — |
| kitx | 标识与校验——uidgen（Snowflake 全局单例+Extract 元数据、手写 RFC 4122 v4 UUID 零第三方依赖、RandomHex/RandomString 拒绝采样随机串）+ validator（go-playground/validator v10 接口化门面：struct tag 校验/自定义规则注册/ValidationError Unwrap 为 BizError 400 对接 middleware.Fail），自 pkg v0.10.0 拆出 | pkg |
| metrics | Prometheus 门面 + 弹性指标族 | core |
| tracing | OpenTelemetry 追踪 + trace_id 真相源 | logx |
| registry | 注册发现核心 + Nacos 实现 + Nacos 配置源 + sdbridge | core、logx |
| data | 数据访问门面——复刻 ekit data/ 能力面：cache/rediscache/crud/distlock/gormx/redisx/timertask/cachetest 八子包 re-export（类型同一，实现住 pkg 与各子仓）+ cached 缓存读写编排（写后失效/事务提交后失效/查询缓存版本号失效/GetByIDs 批量点读 MGet 快路径/bloom 防穿透）+ twolevel 两级缓存与跨实例失效广播（v0.7.0 起本仓唯二非 re-export 实现）；all/ 聚合模块已废弃 retract | concurrency、gormx、logx、pkg、redisx、timertask |
| gormx | SQL/GORM 全家——连接池增强、框架无关 CRUD Service、仓储（原 oekit/gorm，v0.3.0 更名） | logx、pkg |
| redisx | Redis 全家——客户端封装（根包）、分布式缓存 cache/、分布式锁 distlock、分布式限流 ratelimit（Lua 三态 + 故障本地兜底） | concurrency、logx、pkg |
| etcdx | etcd 全家——客户端封装（根包：生产默认值/Health 探活/Close(ctx) 幂等）、registry 注册发现实现（SPI 对接 registry/nacos 链路：Lease KeepAlive 注册 + 丢约指数退避自动重注册、prefix Watch 订阅推送、ConfigStore KV 配置监听）、distlock 分布式锁（Session+concurrency.Mutex 公平排队，接口与 redisx/distlock.Locker 同形可互换，可选无 etcd 降级放行） | concurrency、core、logx、registry |
| httpx | HTTP 客户端 + 负载均衡——服务发现/熔断（规范哨兵 ErrCircuitOpen）/重试/指标/OTel 透传/一致性哈希（权重感知虚拟副本） | concurrency、core、logx、metrics、registry、tracing |
| network | 轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具 | concurrency、logx、tracing |
| timertask | 分片定时任务（FNV-1a、心跳续约）；v0.1.5 起可选可靠投递 DeliveryAtLeastOnce（租约 + Ack + 可见性超时；过期租约回扫重投/死信、停机 best-effort 收租，v0.1.7 接入生命周期真正生效） | concurrency、logx |
| middleware | Hertz 自研标准中间件栈 + contrib 桥接 + hmw + SheddingMiddleware 自适应降载；v0.6.0 起 accesslog 凭证路径脱敏（WithAccessLogRedactPathPrefixes）、ratelimit 默认 KeyFunc 去端口（连接桶→地址桶） | concurrency、core、logx、metrics、pkg、registry、tracing |
| k8s | Kubernetes 客户端封装——RESTConfig/Clientset/DynamicClient 构建（显式 kubeconfig → in-cluster 自动 → KUBECONFIG/~/.kube/config 逐级回退，QPS/Burst 限流默认 50/100）+ leaderelection 子包 Lease 选主简化封装（OnStartedLeading/OnStoppedLeading/OnNewLeader 回调，主备单活，仅要求 Lease CRUD 权限） | logx |
| nns | NNS 服务治理客户端——provider 无关发现/配置/实例缓存 | concurrency、core、logx、registry |
| nns-endpoint | NNS 端点工厂——per-service 熔断、换实例重试、gRPC 池 | concurrency、core、httpx、logx、metrics、nns、registry |
| bootstrap | 应用生命周期引导——bootstrap.Run 骨架 + healthz | concurrency、logx |
| hertz | Hertz 运行时集成——日志桥接、生命周期、SSE/health/metrics、适配器、OTel、crudh/validatorh（基于 kitx/validator）/versionh（原 handler 微库收编） | bootstrap、concurrency、core、gormx、kitx、logx、metrics、middleware、nns、pkg、registry、tracing |
| kitex | Kitex 生态——日志桥接、治理热更、生命周期、BizError 桥接、kitexotel、istio（xDS 动态熔断，原 istio 微库收编） | bootstrap、concurrency、core、logx、metrics、nns、pkg、registry |
| workflow | 组合式流程编排——串行/并行/异步/条件分支 + Hook AOP | concurrency、logx |
| kitex-examples | Kitex echo 示例——client/server + IDL + 治理配置 | bootstrap、concurrency、core、kitex、logx、metrics、nns、pkg、registry、tracing |
依赖严格自底向上；微库已合并为 pkg，旧微库仓已删除。
本仓在矩阵中的位置与依赖见上表「矩阵内依赖」列。历史沿革：源自 ekit 单体库 （git.enjoye.top/enjoydream/ekit v0.34.0，冻结维护），2026-10 拆分为独立 module。
License
MIT — 见 LICENSE
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