从零开始系统学习 Go 编程

Appearance

错误处理与重试策略 ​

Temporal 提供强大的错误处理和重试机制，帮助构建可靠的工作流。本文档详细介绍错误类型、重试策略配置、心跳机制和 Saga 模式实现。

错误类型 ​

Temporal 定义了多种错误类型，每种类型有不同的处理方式。

错误类型分类 ​

┌─────────────────────────────────────────────────────────────┐
│                     Temporal Error Types                     │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ ApplicationError                                       │  │
│  │ - 业务逻辑错误                                         │  │
│  │ - 可配置是否重试                                       │  │
│  │ - 可携带自定义数据                                     │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ TimeoutError                                           │  │
│  │ - 活动执行超时                                         │  │
│  │ - 工作流执行超时                                       │  │
│  │ - 通常不重试                                           │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ CanceledError                                          │  │
│  │ - 工作流被取消                                         │  │
│  │ - Activity 应处理清理                                  │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ TerminatedError                                        │  │
│  │ - 工作流被终止                                         │  │
│  │ - 无法恢复                                             │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ PanicError                                             │  │
│  │ - Activity/Workflow panic                              │  │
│  │ - 不重试                                               │  │
│  └───────────────────────────────────────────────────────┘  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

ApplicationError ​

ApplicationError 是最常用的错误类型，用于表示业务逻辑错误。

go
package activity

import (
    "context"
    "errors"
    "go.temporal.io/sdk/temporal"
)

func ProcessPayment(ctx context.Context, paymentID string) error {
    // 业务验证失败 - 创建 ApplicationError
    if paymentID == "" {
        return temporal.NewApplicationError(
            "payment ID is required",
            "InvalidInput",  // 错误类型
            true,            // nonRetryable
        )
    }
    
    // 认证失败 - 不可重试的错误
    if !authenticate(ctx) {
        return temporal.NewApplicationError(
            "authentication failed",
            "AuthenticationFailed",
            true,
        )
    }
    
    // 暂时性错误 - 可以重试
    if isServiceUnavailable() {
        return temporal.NewApplicationError(
            "service temporarily unavailable",
            "ServiceUnavailable",
            false,  // 可重试
        )
    }
    
    // 带详细信息的错误
    details := PaymentErrorDetails{
        PaymentID: paymentID,
        Reason:    "insufficient_balance",
        Amount:    100.50,
    }
    return temporal.NewApplicationError(
        "insufficient balance",
        "PaymentFailed",
        false,
        details,  // 错误详情
    )
}

type PaymentErrorDetails struct {
    PaymentID string
    Reason    string
    Amount    float64
}

工作流中处理 ApplicationError ​

go
package workflow

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

func PaymentWorkflow(ctx workflow.Context, paymentID string) error {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    err := workflow.ExecuteActivity(ctx, ProcessPayment, paymentID).Get(ctx, nil)
    if err != nil {
        // 解析错误类型
        var appErr *temporal.ApplicationError
        if errors.As(err, &appErr) {
            // 获取错误类型
            errorType := appErr.Type()
            
            switch errorType {
            case "InvalidInput":
                // 输入错误，不重试
                workflow.GetLogger(ctx).Error("Invalid input", "error", appErr)
                return appErr
                
            case "AuthenticationFailed":
                // 认证错误，通知用户
                workflow.ExecuteActivity(ctx, NotifyUser, "auth failed").Get(ctx, nil)
                return appErr
                
            case "ServiceUnavailable":
                // 暂时性错误，等待后重试
                workflow.NewTimer(ctx, 30*time.Second).Get(ctx, nil)
                return workflow.ExecuteActivity(ctx, ProcessPayment, paymentID).Get(ctx, nil)
                
            case "PaymentFailed":
                // 获取错误详情
                var details PaymentErrorDetails
                appErr.Details(&details)
                workflow.GetLogger(ctx).Error("Payment failed", 
                    "paymentID", details.PaymentID,
                    "reason", details.Reason,
                    "amount", details.Amount,
                )
                return appErr
            }
        }
        
        return err
    }
    
    return nil
}

TimeoutError ​

TimeoutError 表示活动或工作流执行超时。

go
func TimeoutHandlingWorkflow(ctx workflow.Context, orderID string) error {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        ScheduleToCloseTimeout: 2 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    err := workflow.ExecuteActivity(ctx, LongRunningActivity, orderID).Get(ctx, nil)
    if err != nil {
        var timeoutErr *temporal.TimeoutError
        if errors.As(err, &timeoutErr) {
            // 区分超时类型
            if timeoutErr.TimeoutType() == temporal.TimeoutTypeStartToClose {
                workflow.GetLogger(ctx).Error("Activity execution timeout")
                // 执行降级处理
                return workflow.ExecuteActivity(ctx, FallbackActivity, orderID).Get(ctx, nil)
            }
            
            if timeoutErr.TimeoutType() == temporal.TimeoutTypeScheduleToClose {
                workflow.GetLogger(ctx).Error("Activity scheduling timeout")
                // 任务队列可能有问题
                return workflow.ExecuteActivity(ctx, NotifyAdmin, "queue issue").Get(ctx, nil)
            }
        }
        return err
    }
    
    return nil
}

CanceledError ​

CanceledError 表示工作流或活动被取消。

go
func CancellableActivity(ctx context.Context, orderID string) error {
    logger := activity.GetLogger(ctx)
    
    // 检查取消状态
    for i := 0; i < 100; i++ {
        // 检查是否被取消
        if ctx.Err() == context.Canceled {
            logger.Info("Activity cancelled, cleaning up")
            // 执行清理操作
            cleanup(ctx, orderID)
            return temporal.NewCanceledError("activity cancelled by workflow")
        }
        
        // 执行工作
        processItem(ctx, orderID, i)
        
        // 模拟工作进度
        time.Sleep(100 * time.Millisecond)
    }
    
    return nil
}

func CancellationHandlingWorkflow(ctx workflow.Context, orderID string) error {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        WaitForCancellation: true,  // 等待取消完成
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 启动活动
    future := workflow.ExecuteActivity(ctx, CancellableActivity, orderID)
    
    // 监听取消信号
    cancelChan := workflow.GetSignalChannel(ctx, "cancel")
    
    selector := workflow.NewSelector(ctx)
    
    var cancelled bool
    selector.AddReceive(cancelChan, func(c workflow.ReceiveChannel, more bool) {
        c.Receive(ctx, &cancelled)
        if cancelled {
            // 取消活动
            future.Cancel()
        }
    })
    
    selector.AddFuture(future, func(f workflow.Future) {
        var err error
        f.Get(ctx, &err)
        
        var canceledErr *temporal.CanceledError
        if errors.As(err, &canceledErr) {
            workflow.GetLogger(ctx).Info("Activity was cancelled")
            // 执行取消后的清理
            workflow.ExecuteActivity(ctx, CleanupOrder, orderID).Get(ctx, nil)
        }
    })
    
    selector.Select(ctx)
    
    return nil
}

重试策略配置 ​

Temporal 提供灵活的重试策略配置，可以针对不同的错误类型设置不同的重试行为。

重试策略参数 ​

参数说明默认值
InitialInterval初始重试间隔1 秒
BackoffCoefficient退避系数（每次重试间隔乘以此值）2.0
MaximumInterval最大重试间隔InitialInterval × 100
MaximumAttempts最大重试次数0（无限）
NonRetryableErrorTypes不重试的错误类型列表空

重试间隔计算 ​

重试间隔计算公式：
Interval(n) = min(MaximumInterval, InitialInterval × BackoffCoefficient^(n-1))

示例（InitialInterval=1s, BackoffCoefficient=2.0, MaximumInterval=1m）：
- 第 1 次重试：1s × 2^0 = 1s
- 第 2 次重试：1s × 2^1 = 2s
- 第 3 次重试：1s × 2^2 = 4s
- 第 4 次重试：1s × 2^3 = 8s
- 第 5 次重试：1s × 2^4 = 16s
- 第 6 次重试：1s × 2^5 = 32s
- 第 7 次重试：min(60s, 1s × 2^6) = min(60s, 64s) = 60s
- 后续重试：保持 60s

配置重试策略 ​

go
package workflow

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

func RetryPolicyWorkflow(ctx workflow.Context, data string) error {
    // 基本重试策略
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    time.Minute,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    return workflow.ExecuteActivity(ctx, RiskyActivity, data).Get(ctx, nil)
}

func AdvancedRetryPolicyWorkflow(ctx workflow.Context, data string) error {
    // 高级重试策略
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    5 * time.Second,    // 初始等待 5 秒
            BackoffCoefficient: 2.0,                // 每次加倍
            MaximumInterval:    5 * time.Minute,   // 最大等待 5 分钟
            MaximumAttempts:    10,                // 最多重试 10 次
            
            // 以下错误类型不会重试
            NonRetryableErrorTypes: []string{
                "InvalidInput",
                "AuthenticationFailed",
                "PermissionDenied",
                "ResourceNotFound",
            },
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    return workflow.ExecuteActivity(ctx, APIActivity, data).Get(ctx, nil)
}

func InfiniteRetryWorkflow(ctx workflow.Context, data string) error {
    // 无限重试（直到成功或超时）
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: time.Hour * 24,  // 24 小时超时
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Minute,
            BackoffCoefficient: 1.5,  // 较小的退避系数
            MaximumInterval:    time.Hour,
            MaximumAttempts:    0,   // 0 表示无限重试
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    return workflow.ExecuteActivity(ctx, CriticalActivity, data).Get(ctx, nil)
}

func QuickRetryWorkflow(ctx workflow.Context, data string) error {
    // 快速重试（适合短暂故障）
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    100 * time.Millisecond,  // 快速开始
            BackoffCoefficient: 1.2,                     // 较小的增长
            MaximumInterval:    5 * time.Second,        // 最多等待 5 秒
            MaximumAttempts:    3,                      // 快速尝试 3 次
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    return workflow.ExecuteActivity(ctx, QuickActivity, data).Get(ctx, nil)
}

不同场景的重试策略 ​

go
func ScenarioBasedRetryWorkflow(ctx workflow.Context, request Request) error {
    logger := workflow.GetLogger(ctx)
    
    // 场景 1: 外部 API 调用 - 中等重试
    apiOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    30 * time.Second,
            MaximumAttempts:    5,
            NonRetryableErrorTypes: []string{
                "InvalidRequest",
                "Unauthorized",
            },
        },
    }
    apiCtx := workflow.WithActivityOptions(ctx, apiOptions)
    
    err := workflow.ExecuteActivity(apiCtx, CallExternalAPI, request).Get(ctx, nil)
    if err != nil {
        logger.Error("API call failed after retries", "error", err)
        return err
    }
    
    // 场景 2: 数据库操作 - 快速重试
    dbOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    200 * time.Millisecond,
            BackoffCoefficient: 1.5,
            MaximumInterval:    2 * time.Second,
            MaximumAttempts:    10,
        },
    }
    dbCtx := workflow.WithActivityOptions(ctx, dbOptions)
    
    err = workflow.ExecuteActivity(dbCtx, UpdateDatabase, request).Get(ctx, nil)
    if err != nil {
        return err
    }
    
    // 场景 3: 文件上传 - 较长重试间隔
    uploadOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    30 * time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    5 * time.Minute,
            MaximumAttempts:    3,
        },
    }
    uploadCtx := workflow.WithActivityOptions(ctx, uploadOptions)
    
    err = workflow.ExecuteActivity(uploadCtx, UploadFile, request).Get(ctx, nil)
    if err != nil {
        // 文件上传失败可以降级处理
        workflow.ExecuteActivity(ctx, SaveToLocalStorage, request).Get(ctx, nil)
    }
    
    return nil
}

心跳与进度报告 ​

心跳机制用于长时间运行的 Activity，报告进度并在 Worker 故障时允许恢复。

心跳的作用 ​

┌─────────────────────────────────────────────────────────────┐
│                    Activity Heartbeat                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌───────────┐     ┌───────────┐     ┌───────────┐         │
│  │ Activity  │────▶│  Server   │────▶│  Heartbeat │         │
│  │ Heartbeat │     │  Heartbeat│     │  Timeout   │         │
│  └───────────┘     │  Deadline │     │  Triggered │         │
│       │            └───────────┘     └───────────┘         │
│       │                   │                    │            │
│       │                   │                    │            │
│       ▼                   ▼                    ▼            │
│  ┌───────────┐     ┌───────────┐     ┌───────────┐         │
│  │  Progress │     │  Heartbeat │     │  Activity │         │
│  │   Saved   │     │   Missed   │     │  Cancelled│         │
│  └───────────┘     └───────────┘     └───────────┘         │
│                                                              │
└─────────────────────────────────────────────────────────────┘

实现心跳 ​

go
package activity

import (
    "context"
    "go.temporal.io/sdk/activity"
)

func LongRunningProcessing(ctx context.Context, items []string) error {
    logger := activity.GetLogger(ctx)
    
    for i, item := range items {
        // 检查是否被取消
        if ctx.Err() == context.Canceled {
            logger.Info("Activity cancelled")
            return activity.ErrCanceled
        }
        
        // 发送心跳并报告进度
        heartbeatDetails := ProcessingProgress{
            CurrentIndex: i,
            TotalItems:   len(items),
            CurrentItem:  item,
            Percentage:   (i + 1) * 100 / len(items),
        }
        activity.GetHeartbeatDetails(ctx, &heartbeatDetails)  // 获取上次心跳详情
        
        // 发送心跳
        activity.Heartbeat(ctx, heartbeatDetails)
        
        // 处理当前项
        err := processItem(ctx, item)
        if err != nil {
            return err
        }
        
        logger.Info("Processed item", 
            "index", i, 
            "progress", heartbeatDetails.Percentage)
    }
    
    return nil
}

type ProcessingProgress struct {
    CurrentIndex int
    TotalItems   int
    CurrentItem  string
    Percentage   int
}

// 支持恢复的长时间 Activity
func ResumableProcessing(ctx context.Context, items []string) error {
    logger := activity.GetLogger(ctx)
    
    // 获取上次心跳详情，用于恢复
    var progress ProcessingProgress
    if activity.HasHeartbeatDetails(ctx) {
        activity.GetHeartbeatDetails(ctx, &progress)
        logger.Info("Resuming from last heartbeat", 
            "startIndex", progress.CurrentIndex + 1)
    } else {
        // 新执行，从头开始
        progress = ProcessingProgress{
            CurrentIndex: -1,
            TotalItems:   len(items),
        }
    }
    
    // 从上次位置继续
    startIndex := progress.CurrentIndex + 1
    for i := startIndex; i < len(items); i++ {
        if ctx.Err() == context.Canceled {
            return activity.ErrCanceled
        }
        
        // 更新进度
        progress.CurrentIndex = i
        progress.CurrentItem = items[i]
        progress.Percentage = (i + 1) * 100 / len(items)
        
        // 发送心跳
        activity.Heartbeat(ctx, progress)
        
        // 处理项
        err := processItem(ctx, items[i])
        if err != nil {
            return err
        }
    }
    
    return nil
}

配置心跳超时 ​

go
package workflow

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

func HeartbeatWorkflow(ctx workflow.Context, items []string) error {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: time.Hour,
        HeartbeatTimeout:    30 * time.Second,  // 心跳超时
        
        // 重试策略：Worker 故障后在其他 Worker 上恢复
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts: 5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    return workflow.ExecuteActivity(ctx, ResumableProcessing, items).Get(ctx, nil)
}

查询 Activity 进度 ​

go
func ProgressMonitoringWorkflow(ctx workflow.Context, items []string) error {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: time.Hour,
        HeartbeatTimeout:    30 * time.Second,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 设置查询处理器，返回活动进度
    var activityProgress ProcessingProgress
    workflow.SetQueryHandler(ctx, "get-progress", func() (ProcessingProgress, error) {
        return activityProgress, nil
    })
    
    // 启动活动
    future := workflow.ExecuteActivity(ctx, LongRunningProcessing, items)
    
    // 定期查询进度
    for !future.IsReady() {
        workflow.NewTimer(ctx, 5*time.Second).Get(ctx, nil)
        
        // 获取心跳详情（需要通过 Activity 的回调获取）
        // 注意：这里只是示例，实际需要在 Activity 中报告进度
    }
    
    return future.Get(ctx, nil)
}

Saga 模式实现 ​

Saga 模式用于管理分布式事务，在某个步骤失败时执行补偿操作。

Saga 模式原理 ​

┌─────────────────────────────────────────────────────────────┐
│                      Saga Pattern                            │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  正向执行：                                                   │
│  ┌────────┐    ┌────────┐    ┌────────┐    ┌────────┐     │
│  │ Step 1 │───▶│ Step 2 │───▶│ Step 3 │───▶│ Step 4 │     │
│  │Create  │    │ Reserve│    │ Payment│    │  Ship  │     │
│  └────────┘    └────────┘    └────────┘    └────────┘     │
│       ✓            ✓            ✓            ✗            │
│                                                              │
│  补偿执行（逆向）：                                            │
│  ┌────────┐    ┌────────┐    ┌────────┐                     │
│  │ Cancel │◀───│ Refund │◀───│Release │                     │
│  │ Order  │    │Payment │    │Inventory│                    │
│  └────────┘    └────────┘    └────────┘                     │
│                                                              │
└─────────────────────────────────────────────────────────────┘

基本 Saga 实现 ​

go
package workflow

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

func OrderSagaWorkflow(ctx workflow.Context, order Order) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 步骤 1: 创建订单
    logger.Info("Step 1: Creating order")
    var orderID string
    err := workflow.ExecuteActivity(ctx, CreateOrder, order).Get(ctx, &orderID)
    if err != nil {
        return err
    }
    
    // 步骤 2: 预留库存
    logger.Info("Step 2: Reserving inventory")
    err = workflow.ExecuteActivity(ctx, ReserveInventory, orderID).Get(ctx, nil)
    if err != nil {
        // 补偿：取消订单
        logger.Info("Compensating: Canceling order")
        workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
        return err
    }
    
    // 步骤 3: 处理支付
    logger.Info("Step 3: Processing payment")
    var paymentID string
    err = workflow.ExecuteActivity(ctx, ProcessPayment, orderID).Get(ctx, &paymentID)
    if err != nil {
        // 补偿：释放库存 -> 取消订单
        logger.Info("Compensating: Releasing inventory")
        workflow.ExecuteActivity(ctx, ReleaseInventory, orderID).Get(ctx, nil)
        
        logger.Info("Compensating: Canceling order")
        workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
        return err
    }
    
    // 步骤 4: 发货
    logger.Info("Step 4: Shipping order")
    err = workflow.ExecuteActivity(ctx, ShipOrder, orderID).Get(ctx, nil)
    if err != nil {
        // 补偿：退款 -> 释放库存 -> 取消订单
        logger.Info("Compensating: Refunding payment")
        workflow.ExecuteActivity(ctx, RefundPayment, paymentID).Get(ctx, nil)
        
        logger.Info("Compensating: Releasing inventory")
        workflow.ExecuteActivity(ctx, ReleaseInventory, orderID).Get(ctx, nil)
        
        logger.Info("Compensating: Canceling order")
        workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
        return err
    }
    
    logger.Info("Order completed successfully", "orderID", orderID)
    return nil
}

高级 Saga 模式 ​

go
package workflow

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

type SagaStep struct {
    Name       string
    Execute    interface{}
    Compensate interface{}
}

type Saga struct {
    steps     []SagaStep
    executed  []int
    compensating bool
}

func NewSaga() *Saga {
    return &Saga{
        steps:    []SagaStep{},
        executed: []int{},
    }
}

func (s *Saga) AddStep(name string, execute, compensate interface{}) {
    s.steps = append(s.steps, SagaStep{
        Name:       name,
        Execute:    execute,
        Compensate: compensate,
    })
}

func (s *Saga) Execute(ctx workflow.Context) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumAttempts:    3,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 执行每个步骤
    for i, step := range s.steps {
        logger.Info("Executing step", "name", step.Name)
        
        err := workflow.ExecuteActivity(ctx, step.Execute).Get(ctx, nil)
        if err != nil {
            logger.Error("Step failed", "name", step.Name, "error", err)
            
            // 开始补偿
            s.compensating = true
            s.Compensate(ctx)
            return err
        }
        
        s.executed = append(s.executed, i)
    }
    
    return nil
}

func (s *Saga) Compensate(ctx workflow.Context) {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 逆向执行补偿
    for i := len(s.executed) - 1; i >= 0; i-- {
        stepIndex := s.executed[i]
        step := s.steps[stepIndex]
        
        logger.Info("Compensating step", "name", step.Name)
        
        err := workflow.ExecuteActivity(ctx, step.Compensate).Get(ctx, nil)
        if err != nil {
            logger.Error("Compensation failed", "name", step.Name, "error", err)
            // 补偿失败记录，但继续执行其他补偿
        }
    }
}

func AdvancedSagaWorkflow(ctx workflow.Context, order Order) error {
    saga := NewSaga()
    
    // 定义步骤和补偿
    saga.AddStep(
        "create-order",
        CreateOrderActivity,
        CancelOrderActivity,
    )
    
    saga.AddStep(
        "reserve-inventory",
        ReserveInventoryActivity,
        ReleaseInventoryActivity,
    )
    
    saga.AddStep(
        "process-payment",
        ProcessPaymentActivity,
        RefundPaymentActivity,
    )
    
    saga.AddStep(
        "ship-order",
        ShipOrderActivity,
        CancelShipmentActivity,
    )
    
    saga.AddStep(
        "send-notification",
        SendNotificationActivity,
        CancelNotificationActivity,
    )
    
    return saga.Execute(ctx)
}

带并行步骤的 Saga ​

go
func ParallelSagaWorkflow(ctx workflow.Context, order Order) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    // 步骤 1: 创建订单
    var orderID string
    err := workflow.ExecuteActivity(ctx, CreateOrder, order).Get(ctx, &orderID)
    if err != nil {
        return err
    }
    
    // 步骤 2: 并行预留多个仓库的库存
    futures := []workflow.Future{
        workflow.ExecuteActivity(ctx, ReserveInventory, orderID, "warehouse-1"),
        workflow.ExecuteActivity(ctx, ReserveInventory, orderID, "warehouse-2"),
        workflow.ExecuteActivity(ctx, ReserveInventory, orderID, "warehouse-3"),
    }
    
    var reserveErrors []error
    for _, future := range futures {
        var err error
        future.Get(ctx, &err)
        if err != nil {
            reserveErrors = append(reserveErrors, err)
        }
    }
    
    if len(reserveErrors) > 0 {
        // 补偿：取消订单
        workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
        return reserveErrors[0]
    }
    
    // 步骤 3: 处理支付
    var paymentID string
    err = workflow.ExecuteActivity(ctx, ProcessPayment, orderID).Get(ctx, &paymentID)
    if err != nil {
        // 补偿：并行释放库存
        for _, warehouse := range []string{"warehouse-1", "warehouse-2", "warehouse-3"} {
            workflow.ExecuteActivity(ctx, ReleaseInventory, orderID, warehouse)
        }
        
        workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
        return err
    }
    
    return nil
}

超时配置最佳实践 ​

超时类型 ​

超时类型说明适用场景
ScheduleToClose从调度到完成的超时活动整体执行时间限制
ScheduleToStart从调度到开始执行的超时任务队列延迟限制
StartToClose从开始执行到完成的超时实际执行时间限制
HeartbeatTimeout心跳超时长时间运行的活动

超时配置示例 ​

go
func TimeoutBestPracticesWorkflow(ctx workflow.Context, request Request) error {
    // 场景 1: 快速 API 调用
    apiOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,  // API 响应通常很快
    }
    
    // 场景 2: 数据库操作
    dbOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Second,  // 数据库操作应快速完成
        ScheduleToCloseTimeout: 30 * time.Second,  // 包含调度延迟
    }
    
    // 场景 3: 文件处理
    fileOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,  // 文件处理可能较慢
        ScheduleToCloseTimeout: 15 * time.Minute,
        HeartbeatTimeout: 30 * time.Second,  // 需要心跳
    }
    
    // 场景 4: 外部服务调用（不可靠）
    externalOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 1 * time.Minute,  // 单次调用超时
        ScheduleToCloseTimeout: 5 * time.Minute,  // 允许多次重试
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts: 3,
        },
    }
    
    // 场景 5: 长时间批处理
    batchOptions := workflow.ActivityOptions{
        StartToCloseTimeout: 1 * time.Hour,
        HeartbeatTimeout: 1 * time.Minute,  // 长心跳间隔
    }
    
    return nil
}

超时计算建议 ​

go
func CalculateTimeouts(activityType string, estimatedDuration time.Duration) workflow.ActivityOptions {
    switch activityType {
    case "api":
        return workflow.ActivityOptions{
            StartToCloseTimeout: estimatedDuration * 3,  // 3倍预估时间
        }
        
    case "database":
        return workflow.ActivityOptions{
            StartToCloseTimeout: estimatedDuration * 2,
            ScheduleToCloseTimeout: estimatedDuration * 5,
        }
        
    case "file-processing":
        return workflow.ActivityOptions{
            StartToCloseTimeout: estimatedDuration * 5,
            ScheduleToCloseTimeout: estimatedDuration * 10,
            HeartbeatTimeout: estimatedDuration / 10,
        }
        
    case "external-service":
        return workflow.ActivityOptions{
            StartToCloseTimeout: estimatedDuration * 2,
            ScheduleToCloseTimeout: estimatedDuration * 10,  // 允许多次重试
        }
        
    default:
        return workflow.ActivityOptions{
            StartToCloseTimeout: estimatedDuration * 3,
        }
    }
}

最佳实践 ​

错误处理最佳实践 ​

go
func RobustWorkflow(ctx workflow.Context, input string) error {
    logger := workflow.GetLogger(ctx)
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            MaximumAttempts:    3,
            NonRetryableErrorTypes: []string{
                "InvalidInput",
                "BusinessRuleViolation",
            },
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    err := workflow.ExecuteActivity(ctx, BusinessActivity, input).Get(ctx, nil)
    if err != nil {
        var appErr *temporal.ApplicationError
        if errors.As(err, &appErr) {
            switch appErr.Type() {
            case "InvalidInput":
                // 业务错误，直接返回
                return appErr
            case "TemporaryFailure":
                // 暂时性错误，尝试降级
                return workflow.ExecuteActivity(ctx, FallbackActivity, input).Get(ctx, nil)
            }
        }
        
        var timeoutErr *temporal.TimeoutError
        if errors.As(err, &timeoutErr) {
            logger.Error("Activity timeout", "type", timeoutErr.TimeoutType())
            return handleTimeout(ctx, input)
        }
        
        return err
    }
    
    return nil
}

重试策略最佳实践 ​

go
// 1. 根据业务场景选择合适的重试次数
// - 关键业务：多次重试（10+）
// - 一般业务：适中重试（3-5）
// - 快速操作：少重试（2-3）

// 2. 设置合理的退避系数
// - 外部 API：较大系数（2.0+），避免雪崩
// - 内部服务：中等系数（1.5-2.0）
// - 快速操作：小系数（1.2-1.5）

// 3. 明确不可重试的错误
ao := workflow.ActivityOptions{
    RetryPolicy: &temporal.RetryPolicy{
        NonRetryableErrorTypes: []string{
            "InvalidInput",        // 输入错误
            "AuthenticationFailed",// 认证错误
            "PermissionDenied",    // 权限错误
            "ResourceNotFound",    // 资源不存在
            "BusinessRuleViolation", // 业务规则违反
        },
    },
}

// 4. 考虑超时与重试的关系
ao := workflow.ActivityOptions{
    StartToCloseTimeout: 30 * time.Second,  // 单次超时
    ScheduleToCloseTimeout: 2 * time.Minute,  // 总超时（包含重试）
    RetryPolicy: &temporal.RetryPolicy{
        MaximumAttempts: 5,
    },
}

相关资源 ​