Skill Creator 实战：如何设计高可用的技能编排系统

2次阅读

没有评论

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

在微服务架构下构建技能编排系统时，开发者常遇到几个典型问题：

技能版本冲突 ：当多个服务依赖同一技能的不同版本时，容易引发兼容性问题。例如，技能 A 依赖 v1.0 的支付服务，而技能 B 依赖 v2.0，导致部署时版本冲突。
执行上下文传递 ：跨服务的技能调用需要传递上下文信息（如用户 ID、会话状态），传统方式通过 HTTP 头部或消息体传递，容易丢失或篡改。
超时与熔断 ：技能链中某个节点响应缓慢会导致整个链路阻塞，缺乏有效的超时控制和熔断机制。
状态追踪困难 ：分布式环境下，技能执行状态分散在各服务中，难以实现端到端的监控与回滚。

基于消息队列 ：
优点：解耦彻底，吞吐量高
缺点：难以实现复杂依赖关系，缺乏可视化编排能力
基于工作流引擎 ：
优点：支持复杂流程编排，自带状态管理
缺点：学习成本高，性能开销大

我们采用有向无环图（DAG）作为核心模型，主要组件包括：

编排控制器 ：解析 DAG 定义，调度节点执行
技能仓库 ：存储所有注册技能及其元数据
状态存储 ：记录每个执行实例的当前状态
监控看板 ：实时展示执行链路与性能指标

架构数据流：
1. 用户提交技能编排请求
2. 控制器生成执行实例并初始化状态
3. 按照 DAG 顺序异步触发技能节点
4. 各节点执行完成后更新状态存储
5. 监控系统聚合数据并告警

# 技能注册示例
class SkillRegistry:
    def register(self, name: str, version: str, endpoint: str):
        """
        关键设计：1. 使用三级存储结构（name/version/endpoint）2. 通过心跳机制保持活性检测
        """key = f"skills/{name}/{version}"redis.hset(key,"endpoint", endpoint)
        redis.expire(key, HEARTBEAT_TIMEOUT)

# 技能发现示例
def discover_skill(name: str, version: str) -> list:
    """
    采用最终一致性模型：1. 优先返回本地缓存数据
    2. 异步更新注册中心信息
    """cached = local_cache.get(f"{name}:{version}")
    if not cached:
        cached = redis.hgetall(f"skills/{name}/{version}")
        local_cache.set(f"{name}:{version}", cached)
    return cached

// 状态机示例
public class SkillStateMachine {
    private State currentState;

    // 执行带有补偿的事务
    public void executeWithFallback(SkillNode node) {
        try {node.execute();
            this.currentState = State.SUCCEEDED;
        } catch (Exception e) {
            // 补偿事务设计要点：// 1. 必须是幂等操作
            // 2. 需要记录补偿日志
            node.compensate(); 
            this.currentState = State.FAILED;
            auditLog.logCompensation(node);
        }
    }
}