Golang工作流引擎实战

工作流引擎是用于自动化业务流程的核心组件，在Golang中可以通过多种方式实现。以下是一个基于状态机模式的简单工作流引擎实现：

核心概念

工作流引擎主要包含：

• 工作流定义：业务流程的模板
• 工作流实例：具体的执行实例
• 任务节点：流程中的步骤
• 流转条件：节点间的转移规则

基础实现

package main

import (
    "context"
    "fmt"
    "log"
)

// 工作流状态
type WorkflowState string

const (
    StatePending   WorkflowState = "pending"
    StateApproving WorkflowState = "approving"
    StateRejected  WorkflowState = "rejected"
    StateCompleted WorkflowState = "completed"
)

// 工作流定义
type Workflow struct {
    ID      string
    Name    string
    States  map[WorkflowState]*State
    Initial WorkflowState
}

// 状态定义
type State struct {
    Name        WorkflowState
    Transitions []Transition
    Action      func(ctx context.Context, data interface{}) error
}

// 状态转移
type Transition struct {
    From   WorkflowState
    To     WorkflowState
    Condition func(ctx context.Context, data interface{}) bool
}

// 工作流实例
type WorkflowInstance struct {
    ID        string
    Workflow  *Workflow
    Current   WorkflowState
    Data      interface{}
}

// 工作流引擎
type WorkflowEngine struct {
    workflows map[string]*Workflow
}

func NewWorkflowEngine() *WorkflowEngine {
    return &WorkflowEngine{
        workflows: make(map[string]*Workflow),
    }
}

// 注册工作流
func (e *WorkflowEngine) RegisterWorkflow(wf *Workflow) {
    e.workflows[wf.ID] = wf
}

// 创建工作流实例
func (e *WorkflowEngine) CreateInstance(workflowID string, data interface{}) (*WorkflowInstance, error) {
    wf, exists := e.workflows[workflowID]
    if !exists {
        return nil, fmt.Errorf("workflow not found: %s", workflowID)
    }
    
    return &WorkflowInstance{
        ID:       generateID(),
        Workflow: wf,
        Current:  wf.Initial,
        Data:     data,
    }, nil
}

// 执行状态转移
func (e *WorkflowEngine) Execute(ctx context.Context, instance *WorkflowInstance) error {
    state, exists := instance.Workflow.States[instance.Current]
    if !exists {
        return fmt.Errorf("state not found: %s", instance.Current)
    }
    
    // 执行当前状态的动作
    if state.Action != nil {
        if err := state.Action(ctx, instance.Data); err != nil {
            return err
        }
    }
    
    // 检查并执行转移
    for _, transition := range state.Transitions {
        if transition.Condition(ctx, instance.Data) {
            instance.Current = transition.To
            return nil
        }
    }
    
    return nil
}

func generateID() string {
    return "instance_" + fmt.Sprintf("%d", time.Now().UnixNano())
}

使用示例

func main() {
    engine := NewWorkflowEngine()
    
    // 定义审批工作流
    approvalWorkflow := &Workflow{
        ID:   "approval_flow",
        Name: "审批流程",
        Initial: StatePending,
        States: map[WorkflowState]*State{
            StatePending: {
                Name: StatePending,
                Action: func(ctx context.Context, data interface{}) error {
                    fmt.Println("提交审批申请")
                    return nil
                },
                Transitions: []Transition{
                    {
                        From: StatePending,
                        To:   StateApproving,
                        Condition: func(ctx context.Context, data interface{}) bool {
                            // 简单的条件判断
                            return data.(map[string]interface{})["amount"].(float64) < 10000
                        },
                    },
                },
            },
            StateApproving: {
                Name: StateApproving,
                Action: func(ctx context.Context, data interface{}) error {
                    fmt.Println("进入审批环节")
                    return nil
                },
                Transitions: []Transition{
                    {
                        From: StateApproving,
                        To:   StateCompleted,
                        Condition: func(ctx context.Context, data interface{}) bool {
                            return data.(map[string]interface{})["approved"].(bool)
                        },
                    },
                    {
                        From: StateApproving,
                        To:   StateRejected,
                        Condition: func(ctx context.Context, data interface{}) bool {
                            return !data.(map[string]interface{})["approved"].(bool)
                        },
                    },
                },
            },
        },
    }
    
    engine.RegisterWorkflow(approvalWorkflow)
    
    // 创建工作流实例
    instance, err := engine.CreateInstance("approval_flow", map[string]interface{}{
        "amount":   5000.0,
        "approved": true,
    })
    if err != nil {
        log.Fatal(err)
    }
    
    // 执行工作流
    ctx := context.Background()
    for instance.Current != StateCompleted && instance.Current != StateRejected {
        if err := engine.Execute(ctx, instance); err != nil {
            log.Fatal(err)
        }
        fmt.Printf("当前状态: %s\n", instance.Current)
    }
}

生产环境建议

1. 持久化存储：使用数据库存储工作流实例状态
2. 分布式支持：考虑使用消息队列处理状态转移
3. 监控告警：添加指标收集和异常监控
4. 超时处理：为每个状态设置超时机制
5. 版本管理：支持工作流定义的版本控制

这个基础实现可以作为起点，根据具体业务需求进行扩展和优化。