Cadence中Skill调用的新手入门指南：从基础概念到实战避坑

1次阅读

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

Cadence 是 Uber 开源的一款分布式工作流引擎，它解决了微服务架构下长时间运行流程的编排难题。在订单处理、数据流水线等场景中，Skill 调用（即 Activity 执行）是最核心的操作之一。新手常会遇到以下问题：

超时不可控：网络抖动导致 Activity 卡死，拖累整个工作流
重试风暴：因未设幂等键，失败重试引发数据重复处理
并发冲突：多个 Worker 同时处理同一 Activity 造成状态不一致

先通过这张简化架构图理解组件关系：

graph LR
  Client-->Workflow
  Workflow-->|Schedule|Activity
  Activity-->Worker
  Worker-->|Respond|Workflow

关键组件定义：

Workflow：持久化状态机，负责编排 Activity 调用顺序
Activity(Skill)：具体业务逻辑执行单元，如支付操作
Worker：执行 Activity 的进程，可水平扩展

以下示例展示订单折扣计算场景：

// 1. 定义 Activity（Skill）type DiscountActivities struct{}

func (a *DiscountActivities) Calculate(ctx context.Context, orderID string) (float64, error) {
    // 幂等性保障：相同 orderID 只处理一次
    if isProcessed(orderID) {return 0, cadence.NewCanceledError("duplicate order")
    }

    // 模拟耗时操作
    select {case <-time.After(30 * time.Second):
        return applyDiscount(orderID)
    case <-ctx.Done():  // 监听取消信号
        return 0, ctx.Err()}
}

// 2. 注册 Worker
func startWorker() {
    worker := worker.New(
        serviceClient,
        domain,
        taskList,
        worker.Options{ActivityPanicPolicy: worker.BlockWorkflow,  // 熔断策略})

    worker.RegisterActivity(&DiscountActivities{})
    worker.Start()}

// 3. 工作流调用
func OrderWorkflow(ctx workflow.Context, orderID string) error {
    ao := workflow.ActivityOptions{
        ScheduleToStartTimeout: time.Minute,
        StartToCloseTimeout:    time.Minute * 5,
        RetryPolicy: &cadence.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2,
            MaximumInterval:    time.Minute,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    var discount float64
    err := workflow.ExecuteActivity(ctx, "Calculate", orderID).Get(&discount)
    if err != nil {// 处理超时 / 重试逻辑}
    return nil
}