Golang基于Cadence引擎的工作流与活动编排框架

Cadence是一个由Uber开源的分布式工作流编排引擎，而cadence-client是Golang语言与Cadence服务交互的客户端库。下面我将详细介绍如何使用这个库进行工作流和活动的编排。

1. 基本概念

• Workflow(工作流): 定义业务流程的执行逻辑
• Activity(活动): 工作流中执行的具体任务
• Worker(工作者): 执行工作流和活动代码的进程

2. 环境准备

首先需要安装Cadence服务并启动，然后安装Go客户端库：

go get go.uber.org/cadence
go get go.uber.org/yarpc
go get go.uber.org/thriftrw

3. 基本使用示例

3.1 定义活动(Activity)

package activities

import (
	"context"
	"time"
)

// GreetingActivity 定义一个简单的活动
func GreetingActivity(ctx context.Context, name string) (string, error) {
	// 模拟耗时操作
	time.Sleep(2 * time.Second)
	return "Hello " + name + "!", nil
}

3.2 定义工作流(Workflow)

package workflows

import (
	"time"

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

	"yourproject/activities"
)

// GreetingWorkflow 定义一个简单的工作流
func GreetingWorkflow(ctx workflow.Context, name string) (string, error) {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
	}
	ctx = workflow.WithActivityOptions(ctx, ao)

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

	return result, nil
}

3.3 注册工作流和活动

package main

import (
	"log"

	"go.uber.org/cadence/activity"
	"go.uber.org/cadence/worker"
	"go.uber.org/cadence/workflow"

	"yourproject/activities"
	"yourproject/workflows"
)

func init() {
	// 注册工作流
	workflow.Register(workflows.GreetingWorkflow)
	
	// 注册活动
	activity.Register(activities.GreetingActivity)
}

func main() {
	// 配置worker
	workerOptions := worker.Options{
		MetricsScope:          tally.NoopScope,
		Logger:               zap.NewNop(),
		DisableWorkflowWorker: false,
	}

	// 创建worker
	worker := worker.New(
		service.NewClient(serviceClient),
		"your-domain",
		"your-task-list",
		workerOptions,
	)

	// 启动worker
	err := worker.Start()
	if err != nil {
		log.Fatal("Failed to start worker", err)
	}

	// 保持运行
	select {}
}

3.4 启动工作流

package main

import (
	"context"
	"time"

	"go.uber.org/cadence/client"
	"go.uber.org/zap"

	"yourproject/workflows"
)

func startWorkflow() {
	// 创建客户端
	cadenceClient, err := client.NewClient(serviceClient, "your-domain", nil)
	if err != nil {
		panic(err)
	}

	// 工作流选项
	workflowOptions := client.StartWorkflowOptions{
		ID:                           "greeting-workflow",
		TaskList:                     "your-task-list",
		ExecutionStartToCloseTimeout: time.Minute * 10,
	}

	// 启动工作流
	we, err := cadenceClient.StartWorkflow(
		context.Background(),
		workflowOptions,
		workflows.GreetingWorkflow,
		"World",
	)
	if err != nil {
		panic(err)
	}

	// 获取结果
	var result string
	err = cadenceClient.GetWorkflow(context.Background(), we.ID, we.RunID).Get(context.Background(), &result)
	if err != nil {
		panic(err)
	}

	zap.L().Info("Workflow completed", zap.String("Result", result))
}

4. 高级特性

4.1 信号(Signal)

// 工作流中接收信号
func SignalWorkflow(ctx workflow.Context) error {
	ch := workflow.GetSignalChannel(ctx, "signal-name")
	
	var signalData string
	ch.Receive(ctx, &signalData)
	
	// 处理信号数据
	workflow.GetLogger(ctx).Info("Received signal", zap.String("data", signalData))
	return nil
}

// 客户端发送信号
func sendSignal() {
	err := cadenceClient.SignalWorkflow(
		context.Background(),
		"workflow-id",
		"run-id",
		"signal-name",
		"signal-data",
	)
	if err != nil {
		panic(err)
	}
}

4.2 查询(Query)

// 工作流中定义查询处理器
func QueryWorkflow(ctx workflow.Context) error {
	queryType := "state"
	err := workflow.SetQueryHandler(ctx, queryType, func() (string, error) {
		return "current-state", nil
	})
	if err != nil {
		return err
	}
	
	// 工作流逻辑...
	return nil
}

// 客户端查询
func queryWorkflow() {
	var result string
	err := cadenceClient.QueryWorkflow(
		context.Background(),
		"workflow-id",
		"run-id",
		"state",
		&result,
	)
	if err != nil {
		panic(err)
	}
}

5. 最佳实践

1. 幂等性: 活动和信号处理应该是幂等的
2. 超时设置: 合理设置活动和工作流的超时时间
3. 错误处理: 工作流中应妥善处理活动错误
4. 日志记录: 使用workflow.GetLogger记录工作流日志
5. 版本控制: 工作流变更时考虑版本兼容性

6. 总结

Cadence提供了强大的工作流编排能力，通过cadence-client库可以方便地在Golang中实现复杂的业务流程。本文介绍了基本的工作流和活动定义、注册、执行以及信号和查询等高级功能。实际使用时，还需要考虑部署架构、监控和错误处理等生产环境因素。