Go 工作流引擎选型：Temporal、Argo 还是自研调度器

一、工作流引擎选错，重构成本比你想象的高

团队在项目早期随手写了一段 for 循环加 goroutine 的"工作流"，三个月后这段代码膨胀到 2000 行，混着重试逻辑、超时处理、状态持久化、回调通知和错误恢复。你开始思考：是不是该引入一个工作流引擎？

选型的核心矛盾在于：轻量级引擎覆盖不了复杂场景，重量级引擎引入过多复杂度。Temporal 功能完备但运维成本高，Argo Workflows 贴合 Kubernetes 生态但有平台锁定风险，自研调度器灵活但需要持续投入。选择不是追求"最好"，而是找到与团队能力和业务阶段匹配的"最合适"。

几种典型场景下的需求差异：简单的 ETL 流水线需要一个能处理 DAG 依赖和重试的执行器；微服务 Saga 事务需要补偿逻辑和长时间运行的等待状态；ML 训练流水线需要 GPU 资源编排和超参数搜索；批处理任务需要定时触发和失败告警。不同场景对引擎的能力要求差异巨大，一刀切的选型决策必然出错。

二、Temporal、Argo Workflows 与自研的核心差异

三个方案的架构差异决定了它们各自适合的场景。

Temporal的核心优势在于：持久化执行（Workflow 状态持久存储在数据库中，服务重启后自动恢复）、强大的重试策略（指数退避、自定义重试、不可重试错误）、原生支持 Saga 模式（补偿逻辑作为 Workflow 的一部分）、多语言 SDK（Go、Java、TypeScript、Python）。代价是：需要运维独立的 Temporal Server（至少 3 节点 + 数据库）、学习曲线陡峭（Workflow 代码必须是确定性的，有特殊的约束）、调试困难（非确定性错误很难复现）。

Argo Workflows的核心优势在于：Kubernetes 原生（每个步骤是一个 Pod，天然利用 K8s 的调度、资源管理和自动扩缩容）、丰富的模板机制（Steps DAG、循环、条件分支）、制品管理（Workflow 间传递输入输出文件）。代价是：Pod 启动延迟（轻量任务不适合，K8s Pod 调度 + 拉镜像有数秒延迟）、平台绑定（离开 Kubernetes 无法运行）、复杂工作流 YAML 可读性差。

自研调度器的优势：完全可控、零外部依赖、可针对业务高度优化。代价：需要持续投入维护、功能容易跟不上需求演进、缺乏社区生态支持。

三、基于 Temporal 的生产级 Go 工作流实现

package workflow import ( "fmt" "time" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) const ( TaskQueueGPUInference = "gpu-inference-task-queue" MaxRetryAttempts = 3 RetryBackoffBase = 2 * time.Second ) // InferenceInput 推理工作流的输入参数 type InferenceInput struct { ModelName string `json:"model_name"` BatchIDs []string `json:"batch_ids"` Priority int `json:"priority"` // 0: 最高 CallbackURL string `json:"callback_url"` MaxWaitSecs int `json:"max_wait_secs"` } // InferenceOutput 推理工作流的输出 type InferenceOutput struct { SuccessCount int `json:"success_count"` FailedCount int `json:"failed_count"` Results map[string]string `json:"results"` TotalCostMs int64 `json:"total_cost_ms"` } // GPUInferenceWorkflow 定义 GPU 推理工作流编排 func GPUInferenceWorkflow(ctx workflow.Context, input InferenceInput) (*InferenceOutput, error) { logger := workflow.GetLogger(ctx) startTime := workflow.Now(ctx) output := &InferenceOutput{ Results: make(map[string]string), } // 第1步：模型加载（Activity，仅执行一次） activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 5 * time.Minute, RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 1, // 模型加载失败不重试 }, }) var loadResult LoadModelResult err := workflow.ExecuteActivity(activityCtx, LoadModelActivity, LoadModelInput{ ModelName: input.ModelName, }).Get(ctx, &loadResult) if err != nil { return nil, fmt.Errorf("模型加载失败: %w", err) } logger.Info("模型加载成功", "model", input.ModelName) // 第2步：并行批量推理（带重试和超时） batchCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: time.Duration(input.MaxWaitSecs) * time.Second, HeartbeatTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{ InitialInterval: RetryBackoffBase, BackoffCoefficient: 2.0, MaximumInterval: 30 * time.Second, MaximumAttempts: MaxRetryAttempts, // 可重试的错误类型 NonRetryableErrorTypes: []string{"InvalidInputError", "ModelUnavailableError"}, }, }) // 使用 Selector 实现超时控制 batchResults := make(chan InferenceResult, len(input.BatchIDs)) timeoutCtx, cancelHandler := workflow.WithCancel(ctx) for i, batchID := range input.BatchIDs { batchID := batchID // 循环变量捕获 workflow.Go(ctx, func(ctx workflow.Context) { var result InferenceResult err := workflow.ExecuteActivity(batchCtx, RunInferenceActivity, RunInferenceInput{ ModelName: loadResult.ModelPath, BatchID: batchID, }, ).Get(timeoutCtx, &result) if err != nil { logger.Error("批量推理失败", "batchID", batchID, "error", err) result = InferenceResult{BatchID: batchID, Success: false, Error: err.Error()} } batchResults <- result }) // 每批推理之间间隔 100ms 防止 GPU 争抢 if i < len(input.BatchIDs)-1 { _ = workflow.Sleep(ctx, 100*time.Millisecond) } } // 收集结果，取最先完成或超时 for i := 0; i < len(input.BatchIDs); i++ { selector := workflow.NewSelector(ctx) selector.AddReceive(batchResults, func(c workflow.ReceiveChannel, more bool) { var result InferenceResult c.Receive(ctx, &result) output.Results[result.BatchID] = result.Output if result.Success { output.SuccessCount++ } else { output.FailedCount++ } }) selector.Select(ctx) } cancelHandler() // 第3步：结果聚合与回调通知 notifyCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 3, }, }) err = workflow.ExecuteActivity(notifyCtx, NotifyCallbackActivity, NotifyCallbackInput{ CallbackURL: input.CallbackURL, Payload: output, }).Get(ctx, nil) if err != nil { logger.Warn("回调通知失败", "error", err) // 回调失败不影响工作流成功完成，后续异步补发 } output.TotalCostMs = workflow.Now(ctx).Sub(startTime).Milliseconds() logger.Info("工作流完成", "success", output.SuccessCount, "failed", output.FailedCount, "cost_ms", output.TotalCostMs) return output, nil } // LoadModelActivity 模型加载 Activity func LoadModelActivity(ctx context.Context, input LoadModelInput) (*LoadModelResult, error) { // 实际实现：从模型仓库拉取 + GPU 显存加载 if input.ModelName == "" { return nil, temporal.NewApplicationError( "model name is required", "InvalidInputError", nil) } return &LoadModelResult{ ModelPath: fmt.Sprintf("/models/%s/v3", input.ModelName), }, nil } // Activity 输入输出结构体定义 type LoadModelInput struct { ModelName string } type LoadModelResult struct { ModelPath string } type RunInferenceInput struct { ModelName string BatchID string } type InferenceResult struct { BatchID string Output string Success bool Error string } type NotifyCallbackInput struct { CallbackURL string Payload *InferenceOutput }

四、选型决策的关键考量

Temporal 的适用边界：适合有状态的长运行工作流（分钟级到天级）、跨服务 Saga 事务、需要精确重试策略的场景。不适合简单的一次性任务编排或对延迟极度敏感（亚毫秒级）的场景。运维成本是隐藏的最大投入——Temporal Server 的版本升级、数据库维护、Worker 生命周期管理都需要专人负责。

Argo Workflows 的适用边界：适合深度依赖 Kubernetes 的组织、ML 训练流水线、ETL 数据处理。不适合需要跨集群、跨云编排的场景，也不适合需要与外部系统做长时间等待交互（等待人工审批）的场景。一个容易忽略的限制是：Argo Workflows 的每个步骤是一个 Pod，最小调度粒度和启动开销决定了它不适合做细粒度任务编排。

自研的触发条件：当你的业务场景足够特殊（如特定硬件拓扑的调度需求）、团队有足够的平台工程能力、并且预计 6 个月内不会发生大规模的流程变更时，自研是合理的。否则，投入产出比会迅速恶化。

五、总结

工作流引擎选型的核心原则是匹配度而非完备度。Temporal 适合需要强保证的长运行工作流，Argo Workflows 适合以 Kubernetes 为核心的编排场景，自研适用于高度定制化的业务需求。无论选择哪个方案，都应该在接入层抽象出一个统一的 Workflow 接口，以便未来切换引擎时修改成本可控。迁移的最小成本不是技术上的，而是设计时就留好退路。