从零开始系统学习 Go 编程

Appearance

工作流版本控制 ​

Temporal 工作流可以长时间运行（数天、数月甚至数年），在此期间业务逻辑可能发生变化。版本控制机制确保已运行的工作流在代码更新后仍能正确执行。

为什么需要版本控制 ​

问题场景 ​

┌─────────────────────────────────────────────────────────────┐
│                    工作流更新问题                             │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  时间线：                                                     │
│                                                              │
│  Day 1: 工作流启动                                            │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ Workflow V1:                                        │    │
│  │ Step A → Step B → Step C                            │    │
│  └─────────────────────────────────────────────────────┘    │
│       │                                                      │
│       │ 执行到 Step B                                        │
│       ▼                                                      │
│  Day 5: 代码更新为 V2                                         │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ Workflow V2:                                        │    │
│  │ Step A → Step B → Step D → Step C  (新增 Step D)    │    │
│  └─────────────────────────────────────────────────────┘    │
│       │                                                      │
│       │ 问题：                                                │    │
│       │ - 事件历史记录了 V1 的执行路径                        │    │
│       │ - 重放时使用 V2 代码                                  │    │
│       │ - 非确定性错误！                                      │    │
│       ▼                                                      │
│  Day 10: 工作流恢复                                           │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ 重放事件历史：                                       │    │
│  │ Event 1: Step A completed                           │    │
│  │ Event 2: Step B completed                           │    │
│  │ Event 3: ???                                        │    │
│  │ V2 期望 Step D，但历史中没有                         │    │
│  └─────────────────────────────────────────────────────┘    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

非确定性错误的类型 ​

变更类型问题示例
新增 Activity历史中没有该事件在 A 和 B 之间新增 D
删除 Activity历史中有该事件但代码中不存在删除中间步骤
修改顺序事件顺序不匹配将 A→B 改为 B→A
修改参数参数不匹配修改 Activity 参数数量或类型

Patching API ​

Temporal 提供 Patching API 用于安全地修改工作流代码。

GetVersion API ​

go
package workflow

import (
    "go.temporal.io/sdk/workflow"
)

const (
    VersionV1 = 1
    VersionV2 = 2
    VersionV3 = 3
)

func VersionedWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 使用 GetVersion 检查版本
    version := workflow.GetVersion(ctx, "add-validation-step", VersionV1, VersionV2)
    
    // V1: 直接处理订单
    if version == VersionV1 {
        logger.Info("Running V1 logic")
        err := workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil)
        if err != nil {
            return err
        }
    }
    
    // V2: 先验证再处理
    if version == VersionV2 {
        logger.Info("Running V2 logic")
        // 新增的验证步骤
        err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil)
        if err != nil {
            return err
        }
        
        err = workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil)
        if err != nil {
            return err
        }
    }
    
    return nil
}

GetVersion 工作原理 ​

┌─────────────────────────────────────────────────────────────┐
│                    GetVersion 工作原理                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  工作流执行时：                                               │
│                                                              │
│  1. 第一次执行（新工作流）                                     │
│     GetVersion("patch-id", min, max)                        │
│     - 返回 max（当前默认版本）                                │
│     - 记录 ChangeVersion 事件                                │
│                                                              │
│  2. 重放执行（已有工作流）                                     │
│     GetVersion("patch-id", min, max)                        │
│     - 检查事件历史中的 ChangeVersion 事件                    │
│     - 如果有记录，返回记录的版本                              │
│     - 如果没有记录，返回 min                                  │
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ Event History:                                        │  │
│  │ 1: WorkflowExecutionStarted                           │  │
│  │ 2: ChangeVersion (patch-id="add-step", version=1)     │  │
│  │ 3: ActivityTaskScheduled (ProcessOrder)               │  │
│  │ 4: ActivityTaskCompleted                              │  │
│  │ 5: WorkflowExecutionCompleted                         │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

多版本演进 ​

go
func MultiVersionWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 版本演进示例
    // V1: 直接处理
    // V2: 添加验证
    // V3: 添加审批
    
    // Patch 1: 添加验证步骤
    validationVersion := workflow.GetVersion(ctx, "add-validation", VersionV1, VersionV2)
    if validationVersion == VersionV2 {
        err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil)
        if err != nil {
            return err
        }
    }
    
    // Patch 2: 添加审批步骤
    approvalVersion := workflow.GetVersion(ctx, "add-approval", VersionV1, VersionV2)
    if approvalVersion == VersionV2 {
        // 等待审批信号
        approvalChan := workflow.GetSignalChannel(ctx, "approval")
        var approved bool
        approvalChan.Receive(ctx, &approved)
        
        if !approved {
            return fmt.Errorf("order not approved")
        }
    }
    
    // 处理订单（所有版本都有）
    err := workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil)
    if err != nil {
        return err
    }
    
    // Patch 3: 添加发送确认邮件
    notificationVersion := workflow.GetVersion(ctx, "add-notification", VersionV1, VersionV2)
    if notificationVersion == VersionV2 {
        workflow.ExecuteActivity(ctx, SendConfirmation, orderID).Get(ctx, nil)
    }
    
    return nil
}

使用 DefaultVersion ​

go
import "go.temporal.io/sdk/workflow"

func DefaultVersionWorkflow(ctx workflow.Context, orderID string) error {
    // workflow.DefaultVersion 表示工作流创建时的版本
    // 用于从未打补丁的工作流
    
    version := workflow.GetVersion(ctx, "my-patch", workflow.DefaultVersion, VersionV2)
    
    if version == workflow.DefaultVersion {
        // 原始逻辑（未打补丁）
        return workflow.ExecuteActivity(ctx, OriginalActivity, orderID).Get(ctx, nil)
    }
    
    if version == VersionV2 {
        // 新逻辑
        return workflow.ExecuteActivity(ctx, NewActivity, orderID).Get(ctx, nil)
    }
    
    return nil
}

Side Effect ​

Side Effect 用于执行非确定性操作但只执行一次。

go
func SideEffectWorkflow(ctx workflow.Context, input string) error {
    logger := workflow.GetLogger(ctx)
    
    // 使用 Side Effect 执行非确定性操作
    var randomValue int
    workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} {
        // 这个函数只在第一次执行时运行
        // 重放时直接返回之前记录的结果
        return rand.Intn(100)
    }).Get(ctx, &randomValue)
    
    logger.Info("Got random value", "value", randomValue)
    
    // 使用 randomValue 进行确定性操作
    if randomValue > 50 {
        return workflow.ExecuteActivity(ctx, ActivityA, input).Get(ctx, nil)
    } else {
        return workflow.ExecuteActivity(ctx, ActivityB, input).Get(ctx, nil)
    }
}

Mutable Side Effect ​

Mutable Side Effect 允许在重放时修改结果。

go
func MutableSideEffectWorkflow(ctx workflow.Context) error {
    // Mutable Side Effect 允许根据当前状态修改结果
    var config Config
    
    workflow.MutableSideEffect(ctx, "config", 
        func(ctx workflow.Context) interface{} {
            // 第一次执行时获取配置
            return loadConfig()
        },
        func(ctx workflow.Context, value interface{}, state interface{}) (interface{}, error) {
            // 重放时可以修改
            oldConfig := value.(Config)
            currentState := state
            
            // 根据当前状态决定是否更新配置
            if needsUpdate(currentState) {
                return updateConfig(oldConfig), nil
            }
            return oldConfig, nil
        },
    ).Get(ctx, &config)
    
    // 使用配置...
    return nil
}

Worker Versioning ​

Worker Versioning 是更系统化的版本控制方式，允许为不同版本的工作流分配不同的 Worker。

Worker Versioning 原理 ​

┌─────────────────────────────────────────────────────────────┐
│                    Worker Versioning                          │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │                    Temporal Server                     │  │
│  │  ┌─────────────────────────────────────────────────┐  │  │
│  │  │              Version Routing                     │  │  │
│  │  │  - Workflow ID → Version ID mapping             │  │  │
│  │  │  - Task routing based on version                │  │  │
│  │  └─────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────────────────────────────┘  │
│                           │                                  │
│           ┌───────────────┼───────────────┐                 │
│           ▼               ▼               ▼                 │
│  ┌───────────────┐ ┌───────────────┐ ┌───────────────┐     │
│  │  Worker V1    │ │  Worker V2    │ │  Worker V3    │     │
│  │  Build ID: A  │ │  Build ID: B  │ │  Build ID: C  │     │
│  │               │ │               │ │               │     │
│  │  Workflow V1  │ │  Workflow V2  │ │  Workflow V3  │     │
│  │  Code         │ │  Code         │ │  Code         │     │
│  └───────────────┘ └───────────────┘ └───────────────┘     │
│                                                              │
│  新工作流：路由到最新版本 Worker                              │
│  现有工作流：路由到创建时版本的 Worker                        │
│                                                              │
└─────────────────────────────────────────────────────────────┘

配置 Build ID ​

go
package main

import (
    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
)

func main() {
    c, _ := client.Dial(client.Options{})
    defer c.Close()
    
    // 创建带 Build ID 的 Worker
    w := worker.New(c, "my-task-queue", worker.Options{
        BuildID: "v2.0.0",  // 当前代码版本标识
        
        // 使用版本ed任务队列
        UseBuildIDForVersioning: true,
    })
    
    // 注册工作流和活动
    w.RegisterWorkflow(MyWorkflow)
    w.RegisterActivity(MyActivity)
    
    w.Run(worker.InterruptCh())
}

配置版本规则 ​

go
func setupVersioning(c client.Client) error {
    // 为任务队列设置版本规则
    
    // 方式 1: 使用 Operator Service API
    ctx := context.Background()
    
    // 添加 Build ID
    err := c.AddBuildID(ctx, "my-task-queue", client.AddBuildIDOptions{
        BuildID: "v2.0.0",
        MakeDefault: true,  // 设为默认版本（新工作流使用）
    })
    if err != nil {
        return err
    }
    
    // 更新版本规则
    err = c.UpdateWorkerBuildIDCompatibilityRules(ctx, "my-task-queue", 
        client.UpdateWorkerBuildIDCompatibilityRulesOptions{
            Rules: []client.BuildIDRule{
                {
                    BuildID: "v2.0.0",
                    MakeDefault: true,
                },
                {
                    BuildID: "v1.0.0",
                    // 旧版本 Worker 可以继续处理旧工作流
                },
            },
        },
    )
    if err != nil {
        return err
    }
    
    return nil
}

版本兼容性规则 ​

go
func versioningExample() {
    // 版本兼容性规则示例
    
    // 规则 1: 简单版本更新
    rules := []client.BuildIDRule{
        {
            BuildID:     "v3.0.0",
            MakeDefault: true,  // 新工作流使用此版本
        },
        {
            BuildID:     "v2.0.0",  // 旧版本继续支持
        },
        {
            BuildID:     "v1.0.0",
        },
    }
    
    // 规则 2: 版本组（兼容版本）
    rulesWithGroups := []client.BuildIDRule{
        {
            BuildID:     "v3.1.0",
            MakeDefault: true,
        },
        {
            BuildID:     "v3.0.0",  // v3.1 和 v3.0 兼容
        },
        {
            BuildID:     "v2.5.0",
            ForCompatibleBuildIDs: []string{"v2.0.0", "v2.1.0"},  // 兼容组
        },
    }
    
    // 规则 3: 渐进式更新
    progressiveRules := []client.BuildIDRule{
        {
            BuildID:     "v3.0.0",
            MakeDefault: true,
        },
        {
            BuildID:              "v2.0.0",
            ForCompatibleBuildIDs: []string{"v1.0.0"},  // v2 可以处理 v1 的工作流
        },
    }
}

版本控制最佳实践 ​

Patch 命名规范 ​

go
// 使用有意义的 Patch ID
const (
    PatchAddValidation    = "order-validation-added"
    PatchAddApproval      = "approval-step-added"
    PatchChangeSequence   = "activity-sequence-changed"
    PatchAddNotifications = "notifications-added"
)

func NamingBestPracticesWorkflow(ctx workflow.Context, orderID string) error {
    // 好的 Patch ID:
    // - 描述性的名称
    // - 包含变更内容
    // - 易于理解
    
    validationVersion := workflow.GetVersion(ctx, 
        "order-validation-added",  // 好：描述性
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    // 不推荐的 Patch ID:
    // - 数字编号
    // - 通用名称
    // - 无意义缩略
    
    _ = workflow.GetVersion(ctx, "patch-1", workflow.DefaultVersion, VersionV2)  // 不推荐
    _ = workflow.GetVersion(ctx, "update", workflow.DefaultVersion, VersionV2)   // 不推荐
    _ = workflow.GetVersion(ctx, "abc", workflow.DefaultVersion, VersionV2)      // 不推荐
    
    return nil
}

版本迁移策略 ​

go
// 策略 1: 保留旧版本一段时间
func VersionMigrationWorkflow(ctx workflow.Context) error {
    // 允许新旧版本并存
    version := workflow.GetVersion(ctx, "migration-patch", VersionV1, VersionV3)
    
    switch version {
    case VersionV1:
        // 旧逻辑
        return oldLogic(ctx)
    case VersionV2:
        // 中间版本逻辑
        return intermediateLogic(ctx)
    case VersionV3:
        // 新逻辑
        return newLogic(ctx)
    }
    
    return nil
}

// 策略 2: 强制迁移（在所有旧工作流完成后）
// 步骤：
// 1. 等待所有使用旧版本的工作流完成
// 2. 移除旧版本代码
// 3. 更新 GetVersion 的 min 为新版本

func ForcedMigrationWorkflow(ctx workflow.Context) error {
    // 最终版本：移除旧版本支持
    version := workflow.GetVersion(ctx, "migration-patch", VersionV3, VersionV3)
    
    // 只有 VersionV3 逻辑
    return newLogic(ctx)
}

// 策略 3: 使用 continue-as-new 迁移
func ContinueAsNewMigrationWorkflow(ctx workflow.Context, input string, version int) error {
    // 检查是否需要迁移到新版本
    currentVersion := workflow.GetVersion(ctx, "main-version", workflow.DefaultVersion, VersionV2)
    
    if currentVersion < VersionV2 {
        // 使用 continue-as-new 迁移到新版本
        return workflow.NewContinueAsNewError(ctx, 
            ContinueAsNewMigrationWorkflow, 
            input, 
            VersionV2,  // 新版本参数
        )
    }
    
    // 新版本逻辑
    return processWithNewVersion(ctx, input)
}

逐步更新策略 ​

go
func IncrementalUpdateWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 第一步更新：添加验证（可能需要 1-2 周）
    validationVersion := workflow.GetVersion(ctx, 
        "step1-add-validation", 
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    if validationVersion == VersionV2 {
        if err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil); err != nil {
            return err
        }
    }
    
    // 第二步更新：添加审批（可能需要 2-3 周）
    approvalVersion := workflow.GetVersion(ctx, 
        "step2-add-approval", 
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    if approvalVersion == VersionV2 {
        approvalChan := workflow.GetSignalChannel(ctx, "approval")
        var approved bool
        approvalChan.Receive(ctx, &approved)
        if !approved {
            return fmt.Errorf("not approved")
        }
    }
    
    // 处理订单
    if err := workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil); err != nil {
        return err
    }
    
    return nil
}

移除补丁 ​

go
func RemovePatchWorkflow(ctx workflow.Context, orderID string) error {
    // 当所有使用旧版本的工作流完成后，可以移除补丁
    
    // 原始版本（有补丁）
    /*
    validationVersion := workflow.GetVersion(ctx, "add-validation", VersionV1, VersionV2)
    if validationVersion == VersionV1 {
        // 旧逻辑
    } else {
        // 新逻辑（包含验证）
    }
    */
    
    // 移除补丁后的版本
    // 直接使用新逻辑，不需要 GetVersion
    
    if err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil); err != nil {
        return err
    }
    
    if err := workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil); err != nil {
        return err
    }
    
    return nil
}

// 移除补丁的时机：
// 1. 确认所有使用旧版本的工作流已完成
// 2. 检查 Temporal UI 中的工作流历史
// 3. 使用 CLI 查询：
//    temporal workflow list --query 'WorkflowType="MyWorkflow" AND CloseTime < "2024-01-01"'

完整示例 ​

订单处理工作流版本演进 ​

go
package workflow

import (
    "fmt"
    "time"
    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"
)

const (
    OrderWorkflowV1 = 1
    OrderWorkflowV2 = 2
    OrderWorkflowV3 = 3
)

// V1: 简单订单处理
// V2: 添加验证步骤
// V3: 添加审批和通知步骤

func OrderProcessingVersionedWorkflow(ctx workflow.Context, order Order) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts: 3,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // ========== Patch 1: 添加验证步骤 ==========
    
    validationVersion := workflow.GetVersion(ctx, 
        "order-validation-patch", 
        OrderWorkflowV1, 
        OrderWorkflowV2,
    )
    
    if validationVersion == OrderWorkflowV2 {
        logger.Info("Running validation step (V2+)")
        err := workflow.ExecuteActivity(ctx, ValidateOrder, order).Get(ctx, nil)
        if err != nil {
            var appErr *temporal.ApplicationError
            if errors.As(err, &appErr) && appErr.Type() == "InvalidOrder" {
                return appErr
            }
            return err
        }
    }
    
    // ========== Patch 2: 添加审批步骤 ==========
    
    approvalVersion := workflow.GetVersion(ctx, 
        "order-approval-patch", 
        OrderWorkflowV1, 
        OrderWorkflowV2,
    )
    
    if approvalVersion == OrderWorkflowV2 {
        logger.Info("Waiting for approval (V3+)")
        
        // 发送审批请求
        err := workflow.ExecuteActivity(ctx, RequestApproval, order).Get(ctx, nil)
        if err != nil {
            return err
        }
        
        // 等待审批信号
        approvalChan := workflow.GetSignalChannel(ctx, "approval")
        timeout := workflow.NewTimer(ctx, 48*time.Hour)
        
        selector := workflow.NewSelector(ctx)
        var approved bool
        var timeoutReached bool
        
        selector.AddReceive(approvalChan, func(c workflow.ReceiveChannel, more bool) {
            var decision ApprovalDecision
            c.Receive(ctx, &decision)
            approved = decision.Approved
        })
        
        selector.AddFuture(timeout, func(f workflow.Future) {
            timeoutReached = true
        })
        
        selector.Select(ctx)
        
        if timeoutReached {
            return fmt.Errorf("approval timeout")
        }
        
        if !approved {
            return fmt.Errorf("order rejected")
        }
    }
    
    // ========== 处理支付（所有版本） ==========
    
    logger.Info("Processing payment")
    var paymentID string
    err := workflow.ExecuteActivity(ctx, ProcessPayment, order).Get(ctx, &paymentID)
    if err != nil {
        return err
    }
    
    // ========== 处理订单 ==========
    
    logger.Info("Processing order")
    err = workflow.ExecuteActivity(ctx, ProcessOrder, order).Get(ctx, nil)
    if err != nil {
        // 补偿：退款
        workflow.ExecuteActivity(ctx, RefundPayment, paymentID).Get(ctx, nil)
        return err
    }
    
    // ========== Patch 3: 添加通知步骤 ==========
    
    notificationVersion := workflow.GetVersion(ctx, 
        "order-notification-patch", 
        OrderWorkflowV1, 
        OrderWorkflowV2,
    )
    
    if notificationVersion == OrderWorkflowV2 {
        logger.Info("Sending notifications (V3+)")
        
        // 并行发送通知
        futures := []workflow.Future{
            workflow.ExecuteActivity(ctx, SendEmailNotification, order),
            workflow.ExecuteActivity(ctx, SendSMSNotification, order),
        }
        
        for _, future := range futures {
            future.Get(ctx, nil)  // 通知失败不影响主流程
        }
    }
    
    logger.Info("Order completed")
    return nil
}

type ApprovalDecision struct {
    Approved bool
    Approver string
    Reason   string
}

使用 SetQueryHandler 支持版本查询 ​

go
func VersionedQueryWorkflow(ctx workflow.Context, orderID string) error {
    // 记录当前版本信息
    var versionInfo struct {
        ValidationVersion int
        ApprovalVersion   int
        NotificationVersion int
    }
    
    // 设置查询处理器，返回版本信息
    workflow.SetQueryHandler(ctx, "get-version-info", func() (interface{}, error) {
        return versionInfo, nil
    })
    
    // 执行带版本控制的逻辑
    versionInfo.ValidationVersion = workflow.GetVersion(ctx, 
        "validation-patch", 
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    versionInfo.ApprovalVersion = workflow.GetVersion(ctx, 
        "approval-patch", 
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    versionInfo.NotificationVersion = workflow.GetVersion(ctx, 
        "notification-patch", 
        workflow.DefaultVersion, 
        VersionV2,
    )
    
    // ... 工作流逻辑
    
    return nil
}

版本控制决策表 ​

变更类型推荐方式说明
新增 ActivityGetVersion使用版本判断是否执行新步骤
删除 ActivityGetVersion使用版本判断是否跳过步骤
修改 Activity 参数GetVersion使用版本判断参数格式
修改顺序GetVersion使用版本判断执行顺序
新增 Signal无需版本Signal 处理器可以动态添加
新增 Query无需版本Query 处理器可以动态添加
大规模重构Worker Versioning使用独立的 Worker 版本
完全重写Continue-As-New新工作流使用新代码

相关资源 ​