Cadence 工作流中动态添加 Skill 的实践与避坑指南

1次阅读

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

在传统的工作流设计中，我们通常会在编写工作流代码时静态定义所有的步骤和逻辑。这种方式在业务简单且稳定的场景下没有问题，但是当业务需要动态扩展能力时就会遇到诸多挑战。

想象一个电商订单处理工作流，最初可能只需要处理订单创建、支付和发货三个步骤。但随着业务发展，我们可能需要动态添加风控检查、优惠券核销、库存预占等额外的 Skill。如果每次新增功能都要修改工作流定义并重新部署，不仅效率低下，还可能影响正在运行的工作流实例。

在 Cadence 中，我们通常有三种方式来实现动态能力扩展：

修改工作流定义：直接修改代码并重新部署
优点：实现简单直接
缺点：需要重启工作流，无法满足动态性要求
子工作流(Child Workflow)：将新增功能封装为子工作流
优点：逻辑隔离清晰
缺点：增加了管理复杂度，性能开销较大
信号量 (Signal) 机制：通过 Signal 动态注入逻辑
优点：完全动态，无需修改工作流定义
缺点：需要自行处理状态管理和幂等性

经过对比，Signal 机制最适合需要高度动态性的场景。

graph LR
    A[客户端] -->| 发送 Signal| B[Cadence 服务]
    B -->| 触发 | C[Signal 处理器]
    C -->| 更新 | D[Skill 状态机]
    D -->| 影响 | E[工作流决策]

以下以 Go 语言为例展示核心代码：

// 注册 Signal 处理器
func (w *MyWorkflow) RegisterSignalHandlers(ctx workflow.Context) {selector := workflow.NewSelector(ctx)

    // 注册添加 Skill 的 Signal
    workflow.SetSignalHandler(ctx, "AddSkill", func(signal []byte) {
        var skill Skill
        if err := json.Unmarshal(signal, &skill); err != nil {workflow.GetLogger(ctx).Error("解析 Skill 失败", zap.Error(err))
            return
        }

        // 幂等性检查：确保不重复添加相同 Skill
        if !w.skillStateMachine.CanAdd(skill.ID) {return}

        // 更新状态机
        w.skillStateMachine.Add(skill)
    })
}

// Skill 状态机实现
type SkillStateMachine struct {skills map[string]Skill
    mu     sync.Mutex
}

func (sm *SkillStateMachine) Add(skill Skill) {sm.mu.Lock()
    defer sm.mu.Unlock()

    // 必须校验版本号防止陈旧信号
    if existing, ok := sm.skills[skill.ID]; ok && existing.Version >= skill.Version {return}

    sm.skills[skill.ID] = skill
}

Signal 处理延迟 ：Signal 处理是异步的，需要考虑决策任务超时(Decision Task Timeout) 问题。建议：
将长时间运行的 Skill 逻辑放入 Activity
控制单个决策任务中的 Signal 处理数量
批量处理：高频率 Signal 场景下，可以考虑批量处理模式：

// 批量处理 Signal 示例
func (w *MyWorkflow) ProcessBatchedSignals() {var batchedSkills []Skill
    workflow.SetBufferedSignalHandler(ctx, "AddSkill", func(signals [][]byte) {
        for _, s := range signals {
            var skill Skill
            if err := json.Unmarshal(s, &skill); err == nil {batchedSkills = append(batchedSkills, skill)
            }
        }
        w.skillStateMachine.BatchAdd(batchedSkills)
    })
}