Golang跨技术栈事件驱动工作流框架插件Workflow的使用

概述

Workflow是一个分布式事件驱动的工作流框架，可以运行健壮、持久和可扩展的顺序业务逻辑。它使用RoleScheduler将工作分布在多个实例上。

主要特性

• 技术栈无关：支持Kafka、Cassandra、Redis、MongoDB、PostgreSQL、MySQL、RabbitMQ或Reflex
• 基于图(DAG)：通过定义称为"Steps"的小工作单元来设计工作流
• TDD支持：采用测试驱动开发
• 回调：支持从webhook或手动触发回调
• 事件融合：添加事件连接器以消费外部事件流
• 钩子：在工作流运行的核心变化时执行钩子
• 调度：支持标准cron规范
• 超时：设置动态或静态等待时间
• 并行消费者：指定运行多少个步骤消费者
• 消费者管理：优雅关闭所有进程

安装

go get github.com/luno/workflow

适配器

Workflow通过适配器实现技术栈无关性。核心适配器包括：

事件流适配器(EventStreamer)

# Kafka适配器
go get github.com/luno/workflow/adapters/kafkastreamer

# Reflex适配器
go get github.com/luno/workflow/adapters/reflexstreamer

记录存储(RecordStore)

# SQL存储适配器
go get github.com/luno/workflow/adapters/sqlstore

角色调度器(RoleScheduler)

# Rink角色调度器
go get github.com/luno/workflow/adapters/rinkrolescheduler

完整示例

第一步：定义工作流

package usage

import (
	"context"

	"github.com/luno/workflow"
)

type Step int

func (s Step) String() string {
	switch s {
	case StepOne:
		return "One"
	case StepTwo:
		return "Two"
	case StepThree:
		return "Three"
	default:
		return "Unknown"
	}
}

const (
	StepUnknown Step = 0
	StepOne     Step = 1
	StepTwo     Step = 2
	StepThree   Step = 3
)

type MyType struct {
	Field string
}

func Workflow() *workflow.Workflow[MyType, Step] {
	b := workflow.NewBuilder[MyType, Step]("my workflow name")

	b.AddStep(StepOne, func(ctx context.Context, r *workflow.Run[MyType, Step]) (Step, error) {
		r.Object.Field = "Hello,"
		return StepTwo, nil
	}, StepTwo)

	b.AddStep(StepTwo, func(ctx context.Context, r *workflow.Run[MyType, Step]) (Step, error) {
		r.Object.Field += " world!"
		return StepThree, nil
	}, StepThree)

	return b.Build(...)
}

第二步：运行工作流

wf := usage.Workflow()

ctx := context.Background()
wf.Run(ctx)

停止工作流：

wf.Stop()

第三步：触发工作流

foreignID := "82347982374982374"
runID, err := wf.Trigger(ctx, foreignID)
if err != nil {
	...
}

等待工作流完成：

foreignID := "82347982374982374"
runID, err := wf.Trigger(ctx, foreignID)
if err != nil {
	...
}

ctx, cancel := context.WithTimeout(ctx, 10 * time.Second)
defer cancel()

record, err := wf.Await(ctx, foreignID, runID, StepThree)
if err != nil {
	...
}

工作流运行状态

一个工作流运行(Run)可以处于以下状态之一：

配置选项

Workflow提供了多种配置选项：

// 设置并行实例数
b.AddStep(StepOne, ..., StepTwo).WithOptions(
    workflow.ParallelCount(5)
)

// 设置轮询频率
b.AddStep(StepOne, ..., StepTwo).WithOptions(
    workflow.PollingFrequency(10 * time.Second)
)

// 设置错误后退时间
b.AddStep(StepOne, ..., StepTwo).WithOptions(
    workflow.ErrBackOff(5 * time.Minute)
)

// 设置延迟警报
b.AddStep(StepOne, ..., StepTwo).WithOptions(
    workflow.LagAlert(15 * time.Minute),
)

// 设置错误计数暂停阈值
b.AddStep(StepOne, ..., StepTwo).WithOptions(
    workflow.PauseAfterErrCount(3),
)

最佳实践

1. 将复杂业务逻辑分解为小步骤
2. 可用于产生新数据而不仅仅是执行逻辑
3. 工作流变更必须向后兼容
4. 不适合低延迟场景
5. 使用Workflow提供的Prometheus指标进行监控和告警