Go DAG 执行器有向无环图中的并行度和失败处理一、DAG 不是画出来好看的是让并行真正生效的骨架工程师画 DAG 很顺手——用 Draw.io 或 Mermaid 把步骤排成一张图感觉一切都在掌控之中。但把这张图画成代码后真正的问题才开始怎么确定哪些步骤可以并行某个步骤失败后哪些下游应该停止并行度大了会不会压垮下游服务DAG 执行器不是简单的拓扑排序加 goroutine。一个生产级的 DAG 执行器需要处理三件事依赖解析确定当前可执行的节点集合、执行控制限制并发度、支持优先级、失败传播失败节点如何影响其下游。以数据流水线为例一个 ETL 任务包含提取Extract、清洗Cleanse、特征计算Feature、聚合Aggregate四个阶段。其中提取可以是多个数据源并行MySQL、Kafka、S3清洗依赖各自的提取结果特征计算依赖清洗结果聚合依赖所有特征计算结果。这个看似简单的流水线实际上包含了并行分支、串行依赖和汇聚点三种拓扑模式。二、DAG 执行器的架构设计核心组件包括DAG 解析器将用户定义的节点和边转为内部数据结构验证无环性调度器持续检查就绪节点所有依赖已满足分发给 Worker Pool 执行Worker Pool限制最大并发 Goroutine 数量防止下游被压垮失败处理器决定失败后策略——向上游传播重试、向下游传播失败、或按照配置跳过状态管理器维护每个节点的执行状态支持查询和断点恢复三、生产级 DAG 执行器实现package dag import ( context fmt log sync time ) // NodeFunc 节点执行函数类型 type NodeFunc func(ctx context.Context, depsOutput map[string]*NodeResult) (interface{}, error) // FailStrategy 失败后处理策略 type FailStrategy int const ( FailStop FailStrategy iota // 停止所有下游 FailSkip // 跳过当前及受影响的节点 FailContinue // 记录失败但继续用于非关键步骤 ) // RetryPolicy 重试策略 type RetryPolicy struct { MaxRetries int Backoff time.Duration Multiplier float64 // 退避系数 } // Node 定义 DAG 中的一个步骤 type Node struct { Name string Func NodeFunc Deps []string // 依赖节点名称 FailStrategy FailStrategy Retry *RetryPolicy Timeout time.Duration Metadata map[string]string } // NodeResult 节点执行结果 type NodeResult struct { Name string Output interface{} Error error Status NodeStatus Elapsed time.Duration Retries int } type NodeStatus int const ( StatusPending NodeStatus iota StatusRunning StatusSuccess StatusFailed StatusSkipped ) // DAGExecutor DAG 执行器 type DAGExecutor struct { nodes map[string]*Node results map[string]*NodeResult mu sync.Mutex maxWorkers int // 控制面 ctx context.Context cancel context.CancelFunc } // NewExecutor 创建 DAG 执行器 func NewExecutor(maxWorkers int) *DAGExecutor { return DAGExecutor{ nodes: make(map[string]*Node), results: make(map[string]*NodeResult), maxWorkers: maxWorkers, } } // AddNode 添加节点在 Execute 前调用 func (e *DAGExecutor) AddNode(node *Node) error { e.mu.Lock() defer e.mu.Unlock() if _, exists : e.nodes[node.Name]; exists { return fmt.Errorf(节点 %s 已存在, node.Name) } e.nodes[node.Name] node e.results[node.Name] NodeResult{Name: node.Name, Status: StatusPending} return nil } // Execute 执行整个 DAG func (e *DAGExecutor) Execute(ctx context.Context) (map[string]*NodeResult, error) { // 验证 DAG 无环 if err : e.validateAcyclic(); err ! nil { return nil, fmt.Errorf(DAG 验证失败: %w, err) } e.ctx, e.cancel context.WithCancel(ctx) defer e.cancel() // 使用带缓冲的通道控制并发度 sem : make(chan struct{}, e.maxWorkers) var wg sync.WaitGroup for { readyNodes : e.getReadyNodes() if len(readyNodes) 0 { // 检查是否所有节点都已完成 remaining, _ : e.countByStatus() if remaining 0 { break } // 有节点未完成但不是就绪状态说明存在依赖链上的失败 // 检查是否有可能继续的节点依赖失败但策略为 Skip/Continue if !e.hasSkippableRemaining() { break } // 短暂等待后重试 select { case -e.ctx.Done(): break case -time.After(100 * time.Millisecond): } continue } for _, nodeName : range readyNodes { wg.Add(1) go func(name string) { defer wg.Done() sem - struct{}{} // 获取信号量 defer func() { -sem }() // 释放信号量 e.executeNode(name) }(nodeName) } wg.Wait() } return e.getResults(), nil } // executeNode 执行单个节点含重试逻辑 func (e *DAGExecutor) executeNode(name string) { e.mu.Lock() node, ok : e.nodes[name] result : e.results[name] if !ok || result.Status StatusRunning { e.mu.Unlock() return } // 检查依赖是否全部成功或允许跳过 depsReady : e.areDepsReady(name) if !depsReady { e.mu.Unlock() return } result.Status StatusRunning e.mu.Unlock() // 收集依赖输出 depsOutput : e.collectDepsOutput(name) // 创建带超时的 context execCtx : e.ctx var execCancel context.CancelFunc if node.Timeout 0 { execCtx, execCancel context.WithTimeout(e.ctx, node.Timeout) defer execCancel() } start : time.Now() output, err : node.Func(execCtx, depsOutput) // 错误处理与重试逻辑 retries : 0 if node.Retry ! nil { for err ! nil retries node.Retry.MaxRetries { retries backoff : node.Retry.Backoff * time.Duration( float64(retries)*node.Retry.Multiplier) log.Printf(节点 %s 第 %d 次重试, 等待 %v, name, retries, backoff) select { case -e.ctx.Done(): e.updateResult(name, nil, e.ctx.Err(), StatusFailed, start, retries) return case -time.After(backoff): } output, err node.Func(execCtx, depsOutput) } } elapsed : time.Since(start) if err ! nil { e.handleNodeFailure(name, err, elapsed, retries) } else { e.updateResult(name, output, nil, StatusSuccess, elapsed, retries) } } // handleNodeFailure 处理节点执行失败 func (e *DAGExecutor) handleNodeFailure(name string, err error, elapsed time.Duration, retries int) { e.mu.Lock() node : e.nodes[name] e.mu.Unlock() switch node.FailStrategy { case FailStop: e.updateResult(name, nil, err, StatusFailed, elapsed, retries) // 取消 context停止所有正在执行的节点 if e.cancel ! nil { e.cancel() } // 传播失败到下游 e.propagateFailureToDownstream(name) case FailSkip: e.updateResult(name, nil, err, StatusSkipped, elapsed, retries) // 跳过模式标记下游节点为 Skipped仅传播给硬依赖失败的下游 // 注意只有直接依赖此节点的下游才受影响 e.propagateSkipToDownstream(name) case FailContinue: // 记录失败但继续执行不受影响的节点 e.updateResult(name, nil, err, StatusFailed, elapsed, retries) } } // propagateFailureToDownstream 向所有下游传播失败 func (e *DAGExecutor) propagateFailureToDownstream(failedNode string) { e.mu.Lock() defer e.mu.Unlock() downstream : e.getDownstreamNodes(failedNode) for _, ds : range downstream { if e.results[ds].Status StatusPending { e.results[ds].Status StatusFailed e.results[ds].Error fmt.Errorf(上游节点 %s 失败, failedNode) e.propagateFailureToDownstream(ds) } } } // propagateSkipToDownstream 跳过模式下标记下游 func (e *DAGExecutor) propagateSkipToDownstream(skippedNode string) { e.mu.Lock() defer e.mu.Unlock() downstream : e.getDownstreamNodes(skippedNode) for _, ds : range downstream { if e.results[ds].Status StatusPending { // 仅当所有依赖都已失败/跳过时才跳过此节点 if e.areAllDepsFailedOrSkipped(ds) { e.results[ds].Status StatusSkipped e.results[ds].Error fmt.Errorf(上游节点 %s 被跳过, skippedNode) e.propagateSkipToDownstream(ds) } } } } // collectDepsOutput 收集当前节点的依赖输出 func (e *DAGExecutor) collectDepsOutput(name string) map[string]*NodeResult { e.mu.Lock() node : e.nodes[name] e.mu.Unlock() output : make(map[string]*NodeResult, len(node.Deps)) e.mu.Lock() for _, dep : range node.Deps { if result, ok : e.results[dep]; ok { output[dep] result } } e.mu.Unlock() return output } // areDepsReady 检查所有依赖是否已就绪 func (e *DAGExecutor) areDepsReady(name string) bool { e.mu.Lock() defer e.mu.Unlock() node : e.nodes[name] for _, dep : range node.Deps { status : e.results[dep].Status // 依赖成功或已跳过Skip 模式才视为就绪 if status ! StatusSuccess { if status StatusSkipped node.FailStrategy ! FailStop { continue // 跳过模式下容忍依赖跳过 } return false } } return true } // areAllDepsFailedOrSkipped 检查节点的所有依赖是否都失败/跳过 func (e *DAGExecutor) areAllDepsFailedOrSkipped(name string) bool { node : e.nodes[name] for _, dep : range node.Deps { status : e.results[dep].Status if status ! StatusFailed status ! StatusSkipped { return false } } return true } // getReadyNodes 获取当前所有就绪节点 func (e *DAGExecutor) getReadyNodes() []string { e.mu.Lock() defer e.mu.Unlock() var ready []string for name, node : range e.nodes { result : e.results[name] if result.Status ! StatusPending { continue } if e.areDepsReady(name) { ready append(ready, name) } } return ready } // getDownstreamNodes 获取某个节点的所有下游节点 func (e *DAGExecutor) getDownstreamNodes(upstream string) []string { var downstream []string for name, node : range e.nodes { for _, dep : range node.Deps { if dep upstream { downstream append(downstream, name) } } } return downstream } // hasSkippableRemaining 检查是否有可跳过的剩余节点 func (e *DAGExecutor) hasSkippableRemaining() bool { for name, result : range e.results { if result.Status StatusPending { node : e.nodes[name] if node.FailStrategy FailSkip || node.FailStrategy FailContinue { return true } } } return false } func (e *DAGExecutor) countByStatus() (pending int, completed int) { e.mu.Lock() defer e.mu.Unlock() for _, result : range e.results { switch result.Status { case StatusPending, StatusRunning: pending default: completed } } return } func (e *DAGExecutor) updateResult(name string, output interface{}, err error, status NodeStatus, elapsed time.Duration, retries int) { e.mu.Lock() defer e.mu.Unlock() e.results[name] NodeResult{ Name: name, Output: output, Error: err, Status: status, Elapsed: elapsed, Retries: retries, } } func (e *DAGExecutor) getResults() map[string]*NodeResult { e.mu.Lock() defer e.mu.Unlock() results : make(map[string]*NodeResult, len(e.results)) for k, v : range e.results { results[k] v } return results } // validateAcyclic 使用 DFS 检测 DAG 是否有环 func (e *DAGExecutor) validateAcyclic() error { visited : make(map[string]int) // 0: 未访问, 1: 访问中, 2: 已完成 var dfs func(string) error dfs func(name string) error { if visited[name] 1 { return fmt.Errorf(检测到环涉及节点 %s, name) } if visited[name] 2 { return nil } visited[name] 1 for _, dep : range e.nodes[name].Deps { if _, exists : e.nodes[dep]; !exists { return fmt.Errorf(节点 %s 依赖不存在的节点 %s, name, dep) } if err : dfs(dep); err ! nil { return err } } visited[name] 2 return nil } for name : range e.nodes { if visited[name] 0 { if err : dfs(name); err ! nil { return err } } } return nil }四、DAG 执行器中的关键权衡并发度上限Worker Pool 的信号量控制最大并发 goroutine 数。这个值需要根据下游系统的承载能力设定——如果是数据库写入步骤并发度过高会压垮连接池如果是纯计算步骤可以设得较高。动态调节并发度根据下游响应延迟自适应是进阶优化方向。失败传播的粒度FailStop 策略停止所有下游是最保守但最安全的做法。FailSkip 跳过受影响的节点但允许不受影响的并行分支继续这在数据流水线中很实用——MySQL 数据源提取失败不应阻塞 Kafka 数据源的后续处理。FailContinue 最激进需要确保下游节点能够处理上游失败产生的缺失数据。重试的幂等性要求DAG 执行器提供了重试机制但重试的前提是节点执行函数必须是幂等的。如果某个节点在写入数据库后失败并重试必须保证不会产生重复记录。这个约束需要由节点实现方保证而不是执行器的职责。五、总结DAG 执行器的核心价值在于将任务间的并行可能性和依赖约束形式化。生产级实现需要处理三个维度依赖解析的准确性、失败传播策略的灵活性、以及并发度的可控性。一个实用的 DAG 执行器不在于功能多全面而在于失败场景下的行为是否可预测——当流水线的某个环节出错时操作者应能明确知道哪些任务会被影响、哪些可以继续。