Golang流式编程库flowgraph使用指南

flowgraph是一个基于Golang的流式数据处理库，它允许开发者以声明式的方式构建数据处理流水线。下面我将详细介绍它的使用方法并提供示例代码。

基本概念

flowgraph的核心概念包括：

• Node：数据处理节点，每个节点执行特定功能
• Edge：连接节点的边，代表数据流向
• Graph：由节点和边组成的数据处理图

安装

go get github.com/vectaport/flowgraph

基本使用示例

示例1：简单数据处理流水线

package main

import (
	"fmt"
	"github.com/vectaport/flowgraph"
)

func main() {
	// 创建新图
	fg := flowgraph.New("simple-example")

	// 创建节点
	src := fg.AddNode("source")
	processor := fg.AddNode("processor")
	sink := fg.AddNode("sink")

	// 定义节点行为
	src.SrcFunc = func() interface{} {
		// 生成数据
		staticData := []int{1, 2, 3, 4, 5}
		for _, v := range staticData {
			return v
		}
		return nil // 结束信号
	}

	processor.ProcFunc = func(data interface{}) interface{} {
		// 处理数据：将整数乘以2
		if num, ok := data.(int); ok {
			return num * 2
		}
		return nil
	}

	sink.SinkFunc = func(data interface{}) {
		// 输出结果
		fmt.Printf("Processed result: %v\n", data)
	}

	// 连接节点
	fg.Connect(src, 0, processor, 0)
	fg.Connect(processor, 0, sink, 0)

	// 运行图
	fg.Run()
}

示例2：多阶段数据处理

package main

import (
	"fmt"
	"github.com/vectaport/flowgraph"
	"strings"
)

func main() {
	fg := flowgraph.New("multi-stage")

	// 创建节点
	generator := fg.AddNode("generator")
	filter := fg.AddNode("filter")
	transformer := fg.AddNode("transformer")
	aggregator := fg.AddNode("aggregator")
	printer := fg.AddNode("printer")

	// 定义节点行为
	words := []string{"apple", "banana", "cherry", "date", "elderberry", "fig"}

	generator.SrcFunc = func() interface{} {
		if len(words) == 0 {
			return nil
		}
		word := words[0]
		words = words[1:]
		return word
	}

	filter.ProcFunc = func(data interface{}) interface{} {
		word := data.(string)
		if len(word) > 4 { // 只保留长度大于4的单词
			return word
		}
		return nil // 过滤掉
	}

	transformer.ProcFunc = func(data interface{}) interface{} {
		word := data.(string)
		return strings.ToUpper(word)
	}

	aggregator.ProcFunc = func(data interface{}) interface{} {
		word := data.(string)
		// 简单聚合：计算字母ASCII码之和
		sum := 0
		for _, c := range word {
			sum += int(c)
		}
		return fmt.Sprintf("%s:%d", word, sum)
	}

	printer.SinkFunc = func(data interface{}) {
		fmt.Println(data)
	}

	// 构建图
	fg.Connect(generator, 0, filter, 0)
	fg.Connect(filter, 0, transformer, 0)
	fg.Connect(transformer, 0, aggregator, 0)
	fg.Connect(aggregator, 0, printer, 0)

	// 运行
	fg.Run()
}

高级特性

并行处理

package main

import (
	"fmt"
	"github.com/vectaport/flowgraph"
	"sync"
	"time"
)

func main() {
	fg := flowgraph.New("parallel")

	src := fg.AddNode("source")
	workers := make([]*flowgraph.Node, 3)
	var wg sync.WaitGroup

	// 数据源
	src.SrcFunc = func() interface{} {
		for i := 0; i < 10; i++ {
			time.Sleep(100 * time.Millisecond)
			return i
		}
		return nil
	}

	// 创建3个工作节点
	for i := range workers {
		workers[i] = fg.AddNode(fmt.Sprintf("worker-%d", i))
		workers[i].ProcFunc = func(data interface{}) interface{} {
			num := data.(int)
			time.Sleep(500 * time.Millisecond) // 模拟耗时处理
			return num * num
		}
		fg.Connect(src, 0, workers[i], 0)
	}

	// 结果收集器
	collector := fg.AddNode("collector")
	collector.SinkFunc = func(data interface{}) {
		fmt.Printf("Result: %v\n", data)
		wg.Done()
	}

	// 连接所有worker到collector
	for _, worker := range workers {
		fg.Connect(worker, 0, collector, 0)
	}

	wg.Add(10) // 预期处理10个数据项
	fg.Run()
	wg.Wait()
}

错误处理

package main

import (
	"errors"
	"fmt"
	"github.com/vectaport/flowgraph"
)

func main() {
	fg := flowgraph.New("error-handling")

	src := fg.AddNode("source")
	processor := fg.AddNode("processor")
	sink := fg.AddNode("sink")

	src.SrcFunc = func() interface{} {
		return "data"
	}

	processor.ProcFunc = func(data interface{}) interface{} {
		str, ok := data.(string)
		if !ok {
			return errors.New("invalid data type")
		}
		if str == "error" {
			return errors.New("simulated error")
		}
		return str + "-processed"
	}

	sink.SinkFunc = func(data interface{}) {
		switch v := data.(type) {
		case error:
			fmt.Printf("Error occurred: %v\n", v)
		default:
			fmt.Printf("Success: %v\n", v)
		}
	}

	fg.Connect(src, 0, processor, 0)
	fg.Connect(processor, 0, sink, 0)

	fg.Run()
}

最佳实践

1. 节点职责单一：每个节点应该只做一件事情
2. 错误处理：在关键节点添加错误处理逻辑
3. 性能考虑：对于计算密集型任务，考虑使用并行节点
4. 资源清理：对于需要清理的资源，可以在节点中添加关闭逻辑

flowgraph提供了一种优雅的方式来构建复杂的数据处理流水线，特别适合ETL、数据转换和实时数据处理等场景。通过组合简单的节点，可以构建出强大的数据处理系统。