Go-TaskFlow 并行任务框架使用指南

Go-TaskFlow 是一个轻量级的 Golang 通用任务并行编程框架，它提供了简单易用的 API 来管理和执行并行任务。下面我将介绍其核心功能和使用方法。

1. 安装

go get github.com/go-taskflow/taskflow

2. 基本概念

• Task: 表示一个可执行的任务单元
• Flow: 表示一组有依赖关系的任务集合
• Executor: 负责执行任务和流程

3. 基本使用示例

3.1 创建简单任务

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/go-taskflow/taskflow"
)

func main() {
	// 创建执行器
	executor := taskflow.NewExecutor()

	// 定义任务
	task1 := taskflow.NewTask("task1", func(ctx context.Context) (interface{}, error) {
		time.Sleep(1 * time.Second)
		return "task1 result", nil
	})

	task2 := taskflow.NewTask("task2", func(ctx context.Context) (interface{}, error) {
		time.Sleep(2 * time.Second)
		return 42, nil
	})

	// 执行任务
	result1, err := executor.Execute(context.Background(), task1)
	if err != nil {
		fmt.Println("Task1 failed:", err)
	} else {
		fmt.Println("Task1 result:", result1)
	}

	result2, err := executor.Execute(context.Background(), task2)
	if err != nil {
		fmt.Println("Task2 failed:", err)
	} else {
		fmt.Println("Task2 result:", result2)
	}
}

3.2 创建并行任务流

func main() {
	executor := taskflow.NewExecutor()

	// 创建任务
	taskA := taskflow.NewTask("A", func(ctx context.Context) (interface{}, error) {
		time.Sleep(1 * time.Second)
		return "A done", nil
	})

	taskB := taskflow.NewTask("B", func(ctx context.Context) (interface{}, error) {
		time.Sleep(2 * time.Second)
		return "B done", nil
	})

	taskC := taskflow.NewTask("C", func(ctx context.Context) (interface{}, error) {
		time.Sleep(500 * time.Millisecond)
		return "C done", nil
	})

	// 创建流程并设置依赖关系
	flow := taskflow.NewFlow("example_flow")
	flow.Add(taskA)
	flow.Add(taskB)
	flow.Add(taskC)
	
	// 设置C依赖A和B
	flow.DependsOn(taskC, taskA, taskB)

	// 执行流程
	results, err := executor.ExecuteFlow(context.Background(), flow)
	if err != nil {
		fmt.Println("Flow execution failed:", err)
		return
	}

	for taskName, result := range results {
		fmt.Printf("%s: %v\n", taskName, result)
	}
}

4. 高级功能

4.1 带参数的任务

taskWithParams := taskflow.NewTask("with_params", func(ctx context.Context) (interface{}, error) {
	// 从上下文中获取参数
	params := taskflow.ParamsFromContext(ctx)
	name := params.GetString("name")
	age := params.GetInt("age")
	
	return fmt.Sprintf("Hello %s, age %d", name, age), nil
})

// 执行时传递参数
result, err := executor.Execute(
	taskflow.NewContextWithParams(context.Background(), 
		taskflow.Params{
			"name": "Alice",
			"age":  30,
		}),
	taskWithParams,
)

4.2 错误处理和重试

taskWithRetry := taskflow.NewTask("retry_task", func(ctx context.Context) (interface{}, error) {
	// 模拟有时会失败的任务
	if time.Now().Unix()%2 == 0 {
		return nil, fmt.Errorf("random error")
	}
	return "success", nil
}).WithRetry(3, 1*time.Second) // 重试3次，每次间隔1秒

4.3 超时控制

taskWithTimeout := taskflow.NewTask("timeout_task", func(ctx context.Context) (interface{}, error) {
	time.Sleep(2 * time.Second)
	return "done", nil
}).WithTimeout(1 * time.Second) // 设置1秒超时

// 或者对整个流程设置超时
flow.WithTimeout(5 * time.Second)

5. 实际应用示例

// 模拟电商订单处理流程
func main() {
	executor := taskflow.NewExecutor()

	// 定义任务
	validateOrder := taskflow.NewTask("validate", func(ctx context.Context) (interface{}, error) {
		// 验证订单逻辑
		return "Order validated", nil
	})

	checkInventory := taskflow.NewTask("check_inventory", func(ctx context.Context) (interface{}, error) {
		// 检查库存逻辑
		return "Inventory checked", nil
	})

	processPayment := taskflow.NewTask("process_payment", func(ctx context.Context) (interface{}, error) {
		// 处理支付逻辑
		return "Payment processed", nil
	})

	updateInventory := taskflow.NewTask("update_inventory", func(ctx context.Context) (interface{}, error) {
		// 更新库存逻辑
		return "Inventory updated", nil
	})

	sendNotification := taskflow.NewTask("send_notification", func(ctx context.Context) (interface{}, error) {
		// 发送通知逻辑
		return "Notification sent", nil
	})

	// 创建流程
	orderFlow := taskflow.NewFlow("order_processing")
	orderFlow.Add(validateOrder, checkInventory, processPayment, updateInventory, sendNotification)

	// 设置依赖关系
	orderFlow.DependsOn(
		checkInventory,    // 检查库存依赖订单验证
		processPayment,    // 处理支付依赖订单验证
		updateInventory,   // 更新库存依赖库存检查和支付处理
		sendNotification,  // 发送通知依赖所有前面的步骤
	)

	// 执行流程
	results, err := executor.ExecuteFlow(context.Background(), orderFlow)
	if err != nil {
		fmt.Println("Order processing failed:", err)
		return
	}

	for taskName, result := range results {
		fmt.Printf("%s: %v\n", taskName, result)
	}
}

6. 性能调优

// 创建带自定义配置的执行器
executor := taskflow.NewExecutor(
	taskflow.WithMaxParallel(10),  // 最大并行任务数
	taskflow.WithQueueSize(100),   // 任务队列大小
	taskflow.WithTimeout(30*time.Second), // 默认超时时间
)

总结

Go-TaskFlow 提供了以下优势：

1. 简单易用的 API 设计
2. 灵活的任务依赖管理
3. 内置的错误处理和重试机制
4. 超时控制和上下文传递
5. 可配置的执行策略

通过合理设计任务和流程，可以轻松构建复杂的并行处理系统，同时保持代码的清晰和可维护性。