Cadence使用Skill深度解析：如何构建高效可靠的工作流系统

1次阅读

没有评论

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

在分布式系统开发中，工作流管理是一个复杂且具有挑战性的任务。开发者常常面临以下痛点：

状态管理复杂：分布式系统中的状态管理需要跨多个服务和节点同步，容易出现不一致问题。
错误处理困难：分布式环境中的错误（如网络分区、服务不可用）难以预测和处理，恢复机制复杂。
可维护性差：随着业务逻辑的复杂化，工作流代码容易变得难以维护和扩展。
调试困难：分布式系统中的工作流执行路径难以追踪，调试成本高。

这些问题使得构建高可靠、易维护的工作流系统成为一项艰巨的任务。

Cadence 是一个开源的工作流引擎，旨在解决分布式系统中的工作流管理问题。Skill 是 Cadence 的核心功能之一，它提供了一种声明式的方式来定义和管理工作流。

Workflow：工作流的定义，包含一系列步骤（Activities）和逻辑。
Activity：工作流中的具体任务，通常是可重试的、幂等的操作。
Worker：执行工作流和 Activity 的进程。
Domain：工作流的逻辑分组，用于隔离不同业务场景的工作流。

工作流定义：开发者通过代码定义工作流的逻辑和步骤。
工作流启动：客户端通过 Cadence 服务启动工作流实例。
任务分发：Cadence 服务将工作流任务分发给可用的 Worker。
任务执行：Worker 执行工作流和 Activity，并将结果返回给 Cadence 服务。
状态持久化：Cadence 服务持久化工作流状态，确保故障恢复时状态不丢失。

适用场景：Airflow 更适合批处理任务调度，而 Cadence 更适合长时间运行的、有状态的工作流。
状态管理：Airflow 的状态管理较为简单，Cadence 提供了更强大的状态持久化和恢复机制。
错误处理：Cadence 内置了重试和超时机制，更适合处理分布式环境中的错误。

血缘关系：Temporal 是 Cadence 的一个分支，两者在核心功能上非常相似。
社区支持：Cadence 有更成熟的社区和更广泛的生产环境使用案例。
功能差异：Temporal 在某些高级功能（如跨集群复制）上有所增强。

安装 Cadence 服务：

docker run --rm -p 7933:7933 -p 7934:7934 -p 7935:7935 ubercadence/local:latest

配置客户端：

import "go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"

func newWorkflowClient() workflowserviceclient.Interface {
    client, err := client.NewClient(service.ClientOptions{
        HostPort:     "127.0.0.1:7933",
        Domain:       "example-domain",
    })
    if err != nil {panic(err)
    }
    return client
}

package main

import (
    "context"
    "go.uber.org/cadence/activity"
    "go.uber.org/cadence/workflow"
    "go.uber.org/zap"
)

func MyWorkflow(ctx workflow.Context, name string) (string, error) {
    // 定义工作流选项
    ao := workflow.ActivityOptions{
        ScheduleToStartTimeout: time.Minute,
        StartToCloseTimeout:    time.Minute,
        HeartbeatTimeout:       time.Second * 20,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    // 执行 Activity
    var result string
    err := workflow.ExecuteActivity(ctx, MyActivity, name).Get(ctx, &result)
    if err != nil {return "", err}

    return result, nil
}

func MyActivity(ctx context.Context, name string) (string, error) {logger := activity.GetLogger(ctx)
    logger.Info("Activity started", zap.String("name", name))
    return "Hello" + name + "!", nil
}

Cadence 提供了多种错误处理机制：

重试策略 ：通过RetryPolicy 配置 Activity 的重试行为。

ao := workflow.ActivityOptions{
    RetryPolicy: &cadence.RetryPolicy{
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaximumInterval:    time.Minute,
        MaximumAttempts:    3,
    },
}

超时控制 ：通过ScheduleToStartTimeout、StartToCloseTimeout 等参数控制 Activity 的超时行为。

心跳机制 ：Activity 可以通过RecordHeartbeat 向 Cadence 服务发送心跳，避免因长时间运行被误判为超时。

func LongRunningActivity(ctx context.Context) error {
    for {
        // 执行任务
        activity.RecordHeartbeat(ctx, "progress")
    }
}

以下是一个完整的工作流示例，包含工作流定义、Activity 实现和错误处理：

package main

import (
    "context"
    "time"

    "go.uber.org/cadence/activity"
    "go.uber.org/cadence/workflow"
    "go.uber.org/zap"
)

func MyWorkflow(ctx workflow.Context, name string) (string, error) {
    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:    3,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    var result string
    err := workflow.ExecuteActivity(ctx, MyActivity, name).Get(ctx, &result)
    if err != nil {workflow.GetLogger(ctx).Error("Activity failed", zap.Error(err))
        return "", err
    }

    return result, nil
}

func MyActivity(ctx context.Context, name string) (string, error) {logger := activity.GetLogger(ctx)
    logger.Info("Activity started", zap.String("name", name))

    // 模拟长时间运行的任务
    for i := 0; i < 5; i++ {activity.RecordHeartbeat(ctx, i)
        time.Sleep(time.Second * 5)
    }

    return "Hello" + name + "!", nil
}