Cadence Skill Language 实战指南：如何设计高可靠的工作流

1次阅读

没有评论

共计 2454 个字符，预计需要花费 7 分钟才能阅读完成。

在分布式系统中构建可靠的工作流一直是开发者面临的重大挑战。工作流通常需要跨多个服务、跨越长时间运行，并且要能够应对各种故障情况。最常见的痛点包括：

状态管理复杂：长时间运行的工作流需要在不同步骤间保持状态，传统方法往往依赖数据库，增加了复杂度
错误恢复困难：网络分区、服务宕机等故障发生时，如何优雅恢复而不重复执行或丢失数据
可观察性差：难以跟踪长时间运行工作流的进度和状态

Cadence 提供了一套专门用于构建可靠分布式工作流的 DSL，其核心概念包括：

工作流是 Cadence 中的核心执行单元，它定义了业务逻辑的执行顺序和状态转换。关键特性包括：

自动持久化状态
支持长时间运行（理论上可以无限期）
可暂停和恢复

Activity 是工作流中的具体任务单元，通常是调用外部服务的操作。重要特点：

支持自动重试
超时控制
心跳机制防止卡死

Signal 是工作流外部与工作流通信的机制，用于异步通知工作流状态变化。

下面我们通过一个完整的订单处理工作流示例，展示如何使用 Cadence Skill Language 构建可靠系统。

// 定义订单处理工作流
func OrderProcessingWorkflow(ctx workflow.Context, orderID string) error {
    // 1. 设置工作流选项
    ao := workflow.ActivityOptions{
        ScheduleToStartTimeout: time.Minute,
        StartToCloseTimeout:    time.Minute,
        HeartbeatTimeout:       time.Second * 20,
        RetryPolicy: &cadence.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    time.Minute,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    // 2. 执行库存检查
    var inventoryStatus string
    err := workflow.ExecuteActivity(ctx, CheckInventoryActivity, orderID).Get(ctx, &inventoryStatus)
    if err != nil {return fmt.Errorf("库存检查失败: %v", err)
    }

    // 3. 处理支付
    var paymentResult string
    err = workflow.ExecuteActivity(ctx, ProcessPaymentActivity, orderID).Get(ctx, &paymentResult)
    if err != nil {return fmt.Errorf("支付处理失败: %v", err)
    }

    // 4. 更新订单状态
    err = workflow.ExecuteActivity(ctx, UpdateOrderStatusActivity, orderID, "COMPLETED").Get(ctx, nil)
    if err != nil {return fmt.Errorf("订单状态更新失败: %v", err)
    }

    return nil
}

合理的超时设置对工作流可靠性至关重要：

ScheduleToStartTimeout：从调度到开始执行的最大等待时间
StartToCloseTimeout：Activity 执行的最大总时间
HeartbeatTimeout：两次心跳之间的最大间隔

对于长时间运行的 Activity，心跳机制可以防止被误判为失败：

func ProcessPaymentActivity(ctx context.Context, orderID string) (string, error) {
    // 模拟长时间运行的任务
    for i := 0; i < 10; i++ {
        // 定期发送心跳
        activity.RecordHeartbeat(ctx, fmt.Sprintf("progress-%d", i))
        time.Sleep(time.Second * 5)
    }
    return "SUCCESS", nil
}

所有 Activity 必须设计为幂等的，确保重试不会导致重复操作。常见策略：

使用唯一 ID 标记操作
检查前置状态
使用 CAS（Compare-And-Swap）操作

正确处理 Signal 避免状态不一致：

func OrderWorkflowWithSignal(ctx workflow.Context, orderID string) error {
    // 设置信号通道
    signalChan := workflow.GetSignalChannel(ctx, "cancel-order")

    // 使用 select 处理信号和正常流程
    selector := workflow.NewSelector(ctx)
    selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
        // 处理取消信号
        workflow.ExecuteActivity(ctx, CancelOrderActivity, orderID)
    })

    // ... 其他工作流逻辑
}