动态工作流引擎实战：基于Skill的高效任务编排与调度

2次阅读

没有评论

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

在微服务架构中，业务逻辑往往分散在多个服务中。传统的工作流系统（如 Activiti）采用静态流程定义，每次流程变更都需要重新部署，无法满足快速迭代的需求。我们经常遇到这些问题：

流程变更需要停机发布
任务依赖关系硬编码在配置中
缺乏实时执行状态监控
横向扩展困难

早期我们尝试过 Drools 规则引擎，但发现两个致命问题：

规则复杂度随业务增长呈指数上升
无法支持跨服务的流程编排

最终我们选择了 Skill-Based 架构，核心优势在于：

每个业务能力封装为独立 Skill（技能）
通过 DAG（有向无环图）动态组合 Skills
支持运行时修改流程拓扑

主要包含四大模块：

流程解析器 ：将 JSON/YAML 定义的流程转换为 DAG
技能仓库 ：存储所有可用的 Skill 实现
调度器 ：负责任务队列管理和资源分配
状态机 ：跟踪每个流程实例的执行状态

使用邻接表存储 DAG 结构，关键数据结构：

type Node struct {
    ID       string
    Skill    string
    DependsOn []string // 前置依赖节点
    Timeout  time.Duration
}

type Workflow struct {Nodes map[string]*Node
}

基于 ETCD 的乐观锁方案：

func acquireLock(key string, ttl int) error {resp, err := client.Txn(ctx).
        If(clientv3.Compare(clientv3.Version(key), "=", 0)).
        Then(clientv3.OpPut(key, "locked", clientv3.WithLease(leaseID))).
        Commit()

    if !resp.Succeeded {return errors.New("lock acquired by others")
    }
    return nil
}

通过 Go plugin 机制实现：

func loadSkill(path string) (Skill, error) {plug, err := plugin.Open(path)
    if err != nil {return nil, err}

    sym, err := plug.Lookup("New")
    if err != nil {return nil, err}

    newFunc, ok := sym.(func() Skill)
    if !ok {return nil, errors.New("invalid symbol type")
    }

    return newFunc(), nil}

对于大数据量任务，采用哈希分片：

根据任务 ID 计算哈希值
按 worker 数量取模分配分片
每个分片单独建立处理队列

实现三级回退策略：

首次失败：立即重试（3 次）
持续失败：指数退避（最多 1 小时）
最终失败：进入死信队列人工处理

使用 Kahn 算法进行拓扑排序：

def has_cycle(graph):
    in_degree = {u:0 for u in graph}

    for u in graph:
        for v in graph[u]:
            in_degree[v] += 1

    queue = deque([u for u in in_degree if in_degree[u] == 0])
    count = 0

    while queue:
        u = queue.popleft()
        count += 1

        for v in graph[u]:
            in_degree[v] -= 1
            if in_degree[v] == 0:
                queue.append(v)

    return count != len(graph)

采用语义化版本控制：

Skill 注册时声明兼容版本范围
调度器匹配时检查 major 版本
自动路由到最新 minor 版本

关键指标埋点：

// 记录任务执行时间
start := time.Now()
defer func() {metrics.Observe("task_duration", time.Since(start).Seconds())
}()

// 错误统计
if err != nil {
    metrics.Incr("task_errors", 
        "skill", skillName,
        "error", err.Error())
}

经过半年生产环境验证，核心指标：