dagflow

基于 DAG 的轻量任务流驱动库。

Features

• DAG 校验（空图 / 多 root / 无 root / 环检测）
• 基于拓扑就绪队列的调度（不再使用 sleep 轮询）
• 多分支并发执行
• 多前置依赖全部完成后再触发下游任务
• 支持 context.Context 取消与超时
• 支持错误传播（任务失败后中止执行）
• 支持运行事件回调（可观测性）

Go Version

• go 1.26
• toolchain go1.26.1

Usage

常用方法：

• Add: 添加任务节点至 DagFlow
• Connect: 连接节点间有向关系
• Validate: 校验 DAG 图
• Run: 使用默认配置执行 DAG
• RunContext: 使用 context 和可选执行参数执行 DAG

Job 接口

最小接口：

type JobNode interface {
	Exec()
	Complete()
	Hashcode() interface{}
	IsFinished() bool
	SetFinished(bo bool)
}

可选扩展（用于错误返回和 context 支持）：

type JobNodeWithError interface {
	JobNode
	ExecE() error
	CompleteE() error
}

type JobNodeWithContext interface {
	JobNode
	ExecContext(ctx context.Context) error
	CompleteContext(ctx context.Context) error
}

RunContext 示例

ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()

err := df.RunContext(ctx, &dagflow.RunOptions{
	MaxConcurrency: 4,
	OnEvent: func(ev dagflow.RunEvent) {
		log.Printf("event=%s job=%v err=%v", ev.Type, ev.Job, ev.Err)
	},
})
if err != nil {
	log.Fatalf("run failed: %v", err)
}

Example 1

上图中，明显不是有向无环图，因为存在两个root节点1和2，所以在执行Validate时会报多个root节点的错。

完整用例见：examples/ex1/ex1.go

Example 2

上图中，主要展示的是无根节点的情况，也就是这个是循环的，所以在执行Validate时会报找不到root节点。

完整用例见：examples/ex2/ex2.go

Example 3

上图中，是一个比较简单的有向无环图，在执行Run时，执行顺序会如图所示，依次执行1、2、3。

完整用例见：examples/ex3/ex3.go

Example 4

上图中，是一个比较简单的有向无环图，在执行Run时，当检测到一个节点后有多个分支的情况时，会采用并发的形式去执行后续的节点，所以执行顺序会是先执行1，然后同时执行2和3。

完整用例见：examples/ex4/ex4.go

Example 5

上图中，是一个相对复杂的有向无环图，结合Example 3和Example 4的情况，执行顺序是首先执行1，随后并行执行2、3、4，此时执行完4后，就准备执行6，但检测到要执行6的前提会是先执行完成5，所以后面的顺序是执行完5后再执行6。

完整用例见：examples/ex5/ex5.go

如果这个小工具对你有帮助，欢迎 Star 支持一下 😆