Golang框架实战-KisFlow流式计算框架(4)-数据流

kis-flow源代码：https://github.com/aceld/kis-flow

3.1 数据类型定义

KisFlow中可以传递任意类型数据作为Flow的数据源。而且KisFlow支持批量数据的流逝计算处理。

首先需要对KisFlow中内部支持的数据类型做一个基本的定义，我们将这部分的定义代码写在kis-flow/common/中的data_type.go 文件中。

kis-flow/common/data_type.go

代码语言：go复制

package common

// KisRow 一行数据
type KisRow interface{}

// KisRowArr 一次业务的批量数据
type KisRowArr []KisRow

/*
	KisDataMap 当前Flow承载的全部数据，
   	key	:  数据所在的Function ID
    value: 对应的KisRow
*/
type KisDataMap map[string]KisRowArr

• KisRow ：表示一行数据，可以是任意的数据类型，比如字符串，json字符串，一些序列化的二进制数据， protobuf，yaml字符串等，均可。
• KisRowArr：表示多行数据，也就是一次提交的批量数据，他是KisRow的数组集合。
• KisDataMap ：表示当前Flow承载的全部数据。是一个map[string]KisRowArr类型，其中key为数据所在的Function ID，value为数据。

3.2 KisFlow数据流处理

在KisFlow模块中，新增一些存放数据的成员，如下：

kis-flow/flow/kis_flow.go

代码语言：go复制

// KisFlow 用于贯穿整条流式计算的上下文环境
type KisFlow struct {
	// 基础信息
	Id   string                // Flow的分布式实例ID(用于KisFlow内部区分不同实例)
	Name string                // Flow的可读名称
	Conf *config.KisFlowConfig // Flow配置策略

	// Function列表
	Funcs          map[string]kis.Function // 当前flow拥有的全部管理的全部Function对象, key: FunctionID
	FlowHead       kis.Function            // 当前Flow所拥有的Function列表表头
	FlowTail       kis.Function            // 当前Flow所拥有的Function列表表尾
	flock          sync.RWMutex            // 管理链表插入读写的锁
	ThisFunction   kis.Function            // Flow当前正在执行的KisFunction对象
	ThisFunctionId string                  // 当前执行到的Function ID (策略配置ID)
	PrevFunctionId string                  // 当前执行到的Function 上一层FunctionID(策略配置ID)

	// Function列表参数
	funcParams map[string]config.FParam // flow在当前Function的自定义固定配置参数,Key:function的实例KisID, value:FParam
	fplock     sync.RWMutex             // 管理funcParams的读写锁

	//          数据           
	buffer common.KisRowArr  // 用来临时存放输入字节数据的内部Buf, 一条数据为interface{}, 多条数据为[]interface{} 也就是KisBatch
	data   common.KisDataMap // 流式计算各个层级的数据源
	inPut  common.KisRowArr  // 当前Function的计算输入数据
}

• buffer: 用来临时存放输入字节数据的内部Buf, 一条数据为interface{}, 多条数据为[]interface{} 也就是KisBatch
• data: 流式计算各个层级的数据源
• inPut: 当前Function的计算输入数据

后续章节会使用到这几个成员属性，这里先做为了解。

因为data是一个map类型，所以需要在NewKisFlow()中，对其进行初始化操作：

kis-flow/flow/kis_flow.go

代码语言：go复制

// NewKisFlow 创建一个KisFlow.
func NewKisFlow(conf *config.KisFlowConfig) kis.Flow {
	flow := new(KisFlow)
	// 实例Id
	flow.Id = id.KisID(common.KisIdTypeFlow)

	// 基础信息
	flow.Name = conf.FlowName
	flow.Conf = conf

	// Function列表
	flow.Funcs = make(map[string]kis.Function)
	flow.funcParams = make(map[string]config.FParam)

	//          数据data        
	flow.data = make(common.KisDataMap)

	return flow
}

3.2.2 业务提交数据接口

KisFlow的开发者在编写业务时，可以通过flow实例来进行提交业务源数据，所以我们需要给Flow抽象层新增一个提交数据的接口：

kis-flow/kis/flow.go

代码语言：go复制

package kis

import (
	"context"
	"kis-flow/common"
	"kis-flow/config"
)

type Flow interface {
	// Run 调度Flow，依次调度Flow中的Function并且执行
	Run(ctx context.Context) error
	// Link 将Flow中的Function按照配置文件中的配置进行连接
	Link(fConf *config.KisFuncConfig, fParams config.FParam) error
	// CommitRow         提交Flow数据到即将执行的Function层     
	CommitRow(row interface{}) error
}

新增接口 CommitRow(any interface{}) error。

在kis-flow/flow/kis_flow_data.go中实现KisFlow的该接口。

kis-flow/flow/kis_flow_data.go

代码语言：go复制

func (flow *KisFlow) CommitRow(row interface{}) error {

	flow.buffer = append(flow.buffer, row)

	return nil
}

CommitRow() 为提交Flow数据, 一行数据，如果是批量数据可以提交多次。 所有提交的数据都会暂存在flow.buffer 成员中，作为缓冲区。

3.2.3 KisFlow内部数据提交

现在开发者可以通过CommitRow()将数据提交到buffer中，但是在KisFlow内部需要一个内部接口来将buffer提交到KisFlow的data中，作为之后当前Flow全部Function的上下文数据供使用。所以我们这里需要再提供两个接口。分别是首次提交数据commitSrcData()和中间层提交数据commitCurData()两个函数。

A. 首层数据提交

kis-flow/flow/kis_flow_data.go

代码语言：go复制

// commitSrcData 提交当前Flow的数据源数据, 表示首次提交当前Flow的原始数据源
// 将flow的临时数据buffer，提交到flow的data中,(data为各个Function层级的源数据备份)
// 会清空之前所有的flow数据
func (flow *KisFlow) commitSrcData(ctx context.Context) error {

	// 制作批量数据batch
	dataCnt := len(flow.buffer)
	batch := make(common.KisRowArr, 0, dataCnt)

	for _, row := range flow.buffer {
		batch = append(batch, row)
	}

	// 清空之前所有数据
	flow.clearData(flow.data)

	// 首次提交，记录flow原始数据
	// 因为首次提交，所以PrevFunctionId为FirstVirtual 因为没有上一层Function
	flow.data[common.FunctionIdFirstVirtual] = batch

	// 清空缓冲Buf
	flow.buffer = flow.buffer[0:0]

	log.Logger().DebugFX(ctx, "====> After CommitSrcData, flow_name = %s, flow_id = %snAll Level Data =n % vn", flow.Name, flow.Id, flow.data)

	return nil
}

//ClearData 清空flow所有数据
func (flow *KisFlow) clearData(data common.KisDataMap) {
	for k := range data {
		delete(data, k)
	}
}

实际上commitSrcData()在整个的Flow运行周期只会执行一次，这个作为当前Flow的始祖源数据。

commitSrcData() 的最终目的是 将buffer的数据提交到data[FunctionIdFirstVirtual] 中。 这里要注意的是FunctionIdFirstVirtual是一个虚拟fid，作为所有Function的上游Function ID。 并且首次提交之后，flow.buffer的数据将被清空。

B. 中间层数据提交

kis-flow/flow/kis_flow_data.go

代码语言：go复制

//commitCurData 提交Flow当前执行Function的结果数据
func (flow *KisFlow) commitCurData(ctx context.Context) error {

	//判断本层计算是否有结果数据,如果没有则退出本次Flow Run循环
	if len(flow.buffer) == 0 {
		return nil
	}

	// 制作批量数据batch
	batch := make(common.KisRowArr, 0, len(flow.buffer))

	//如果strBuf为空，则没有添加任何数据
	for _, row := range flow.buffer {
		batch = append(batch, row)
	}

	//将本层计算的缓冲数据提交到本层结果数据中
	flow.data[flow.ThisFunctionId] = batch

	//清空缓冲Buf
	flow.buffer = flow.buffer[0:0]

	log.Logger().DebugFX(ctx, " ====> After commitCurData, flow_name = %s, flow_id = %snAll Level Data =n % vn", flow.Name, flow.Id, flow.data)

	return nil
}

commitCurData()会在每次Function执行计算后，将当前Function的计算结果数据进行提交。 commitCurData() 会在Flow的流式计算过程中被执行多次。

commitCurData()的最终目的是将将buffer的数据提交到data[flow.ThisFunctionId] 中 。ThisFunctionId也就是当前正在执行Function，同时也是下一层将要执行的Function的上一层。

提交之后，flow.buffer的数据将被清空。

3.2.4 获取正在执行Function的源数据

至于每层Function的源数据如何得到，我们可以通过getCurData()方法得到。 通过PrevFunctionId进行索引，因为获取当前Function的源数据，就是上一层Function的结果数据，所以我们通过PrevFunctionId来得到上一层Function的Id，从dataPrevFunctionId 中可以得到数据源。

kis-flow/flow/kis_flow_data.go

代码语言：go复制

// getCurData 获取flow当前Function层级的输入数据
func (flow *KisFlow) getCurData() (common.KisRowArr, error) {
	if flow.PrevFunctionId == "" {
		return nil, errors.New(fmt.Sprintf("flow.PrevFunctionId is not set"))
	}

	if _, ok := flow.data[flow.PrevFunctionId]; !ok {
		return nil, errors.New(fmt.Sprintf("[%s] is not in flow.data", flow.PrevFunctionId))
	}

	return flow.data[flow.PrevFunctionId], nil
}

3.2.5 数据流链式调度处理

下面我们就要在flow.Run()方法中，来加入数据流的处理动作。

kis-flow/flow/kis_flow.go

代码语言：go复制

// Run 启动KisFlow的流式计算, 从起始Function开始执行流
func (flow *KisFlow) Run(ctx context.Context) error {

	var fn kis.Function

	fn = flow.FlowHead

	if flow.Conf.Status == int(common.FlowDisable) {
		//flow被配置关闭
		return nil
	}

    // ========= 数据流 新增 ===========
	// 因为此时还没有执行任何Function, 所以PrevFunctionId为FirstVirtual 因为没有上一层Function
	flow.PrevFunctionId = common.FunctionIdFirstVirtual

	// 提交数据流原始数据
	if err := flow.commitSrcData(ctx); err != nil {
		return err
	}
    // ========= 数据流 新增 ===========


	//流式链式调用
	for fn != nil {

        // ========= 数据流 新增 ===========
		// flow记录当前执行到的Function 标记
		fid := fn.GetId()
		flow.ThisFunction = fn
		flow.ThisFunctionId = fid

		// 得到当前Function要处理与的源数据
		if inputData, err := flow.getCurData(); err != nil {
			log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %sn", err.Error())
			return err
		} else {
			flow.inPut = inputData
		}
        // ========= 数据流 新增 ===========


		if err := fn.Call(ctx, flow); err != nil {
			//Error
			return err
		} else {
			//Success

            // ========= 数据流 新增 ===========
			if err := flow.commitCurData(ctx); err != nil {
				return err
			}

			// 更新上一层FuncitonId游标
			flow.PrevFunctionId = flow.ThisFunctionId
            // ========= 数据流 新增 ===========

			fn = fn.Next()
		}
	}

	return nil
}

• 在run() 刚执行的时候，对PrevFunctionId 进行初始化，设置为 FunctionIdFirstVirtual。
• 在run() 刚执行的时候，执行commitSrcData()将业务赋值的的buffer数据提交到dataFunctionIdFirstVirtual中。
• 进入循环，执行每个Function的时候，getCurData()获取到当前Function的源数据，并且放在flow.inPut 成员中。
• 进入循环，更正ThisFunctionId 游标为当前Function ID。
• 进入循环，每个Funciton执行完毕后，将Function产生的结果数据通过commitCurData()进行提交，并且改变PrevFunctionId为当前FunctionID， 进入下一层。

很显然，我们还需要让Flow给开发者提供一个获取Input数据的接口。

kis-flow/kis/flow.go

代码语言：go复制

package kis

import (
	"context"
	"kis-flow/common"
	"kis-flow/config"
)

type Flow interface {
	// Run 调度Flow，依次调度Flow中的Function并且执行
	Run(ctx context.Context) error
	// Link 将Flow中的Function按照配置文件中的配置进行连接
	Link(fConf *config.KisFuncConfig, fParams config.FParam) error
	// CommitRow 提交Flow数据到即将执行的Function层
	CommitRow(row interface{}) error

    //                       
	// Input 得到flow当前执行Function的输入源数据
	Input() common.KisRowArr
}

实现如下：

kis-flow/flow/kis_flow_data.go

代码语言：go复制

// Input 得到flow当前执行Function的输入源数据
func (flow *KisFlow) Input() common.KisRowArr {
	return flow.inPut
}

3.3 KisFunction的数据流处理

由于我们的Function调度模块还目前还没有实现，所以有关Function在执行Call()方法的时候，只能暂时将业务计算的逻辑写死在KisFlow框架中。 在下一章节，我们会将这部分的计算逻辑开放给开发者进行注册自己的业务。

现在Flow已经将数据传递给了每层的Function，那么在Function中我们下面来简单模拟一下业务的基础计算逻辑。

我们暂时修改KisFunctionC和 KisFunctionE 两个模块的Call()代码.

假设KisFunctionC 是 KisFunctionE的上层。

kis-flow/function/kis_function_c.go

代码语言：go复制

type KisFunctionC struct {
	BaseFunction
}

func (f *KisFunctionC) Call(ctx context.Context, flow kis.Flow) error {
	log.Logger().InfoF("KisFunctionC, flow = % vn", flow)

	//TODO 调用具体的Function执行方法
	//处理业务数据
	for i, row := range flow.Input() {
		fmt.Printf("In KisFunctionC, row = % vn", row)

		// 提交本层计算结果数据
		_ = flow.CommitRow("Data From KisFunctionC, index "   " "   fmt.Sprintf("%d", i))
	}

	return nil
}

kis-flow/function/kis_function_e.gotype KisFunctionE struct { BaseFunction }func (f *KisFunctionE) Call(ctx context.Context, flow kis.Flow) error { log.Logger().InfoF("KisFunctionE, flow = % vn", flow)// TODO 调用具体的Function执行方法 //处理业务数据 for _, row := range flow.Input() { fmt.Printf("In KisFunctionE, row = % vn", row) } return nil}

3.4 数据流单元测试

下面我们模拟一个简单的计算业务，测试下每层的Function是否可以得到数据，并且将计算结果传递给下一层。

kis-flow/test/kis_flow_test.go

代码语言：go复制

func TestNewKisFlowData(t *testing.T) {
	ctx := context.Background()

	// 1. 创建2个KisFunction配置实例
	source1 := config.KisSource{
		Name: "公众号抖音商城户订单数据",
		Must: []string{"order_id", "user_id"},
	}

	source2 := config.KisSource{
		Name: "用户订单错误率",
		Must: []string{"order_id", "user_id"},
	}

	myFuncConfig1 := config.NewFuncConfig("funcName1", common.C, &source1, nil)
	if myFuncConfig1 == nil {
		panic("myFuncConfig1 is nil")
	}

	myFuncConfig2 := config.NewFuncConfig("funcName2", common.E, &source2, nil)
	if myFuncConfig2 == nil {
		panic("myFuncConfig2 is nil")
	}

	// 2. 创建一个 KisFlow 配置实例
	myFlowConfig1 := config.NewFlowConfig("flowName1", common.FlowEnable)

	// 3. 创建一个KisFlow对象
	flow1 := flow.NewKisFlow(myFlowConfig1)

	// 4. 拼接Functioin 到 Flow 上
	if err := flow1.Link(myFuncConfig1, nil); err != nil {
		panic(err)
	}
	if err := flow1.Link(myFuncConfig2, nil); err != nil {
		panic(err)
	}

	// 5. 提交原始数据
	_ = flow1.CommitRow("This is Data1 from Test")
	_ = flow1.CommitRow("This is Data2 from Test")
	_ = flow1.CommitRow("This is Data3 from Test")

	// 6. 执行flow1
	if err := flow1.Run(ctx); err != nil {
		panic(err)
	}
}

这里我们通过flow.CommitRow()提交了3行数据，每行数据是一个字符串，当然数据格式可以任意，数据类型也可以任意，只需要在各层的Function业务自身确定拉齐好即可。

cd到kis-flow/test/下执行命令：

代码语言：bash复制

go test -test.v -test.paniconexit0 -test.run TestNewKisFlowData

结果如下：

代码语言：bash复制

=== RUN   TestNewKisFlowData
context.Background
====> After CommitSrcData, flow_name = flowName1, flow_id = flow-8b607ae6d55048408dae1f4e8f6dca6f
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]

KisFunctionC, flow = &{Id:flow-8b607ae6d55048408dae1f4e8f6dca6f Name:flowName1 Conf:0xc00015a780 Funcs:map[func-2182fa1a049f4c1c9eeb641f5292f09f:0xc0001381e0 func-f3e7d7868f44448fb532935768ea2ca1:0xc000138190] FlowHead:0xc000138190 FlowTail:0xc0001381e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc000138190 ThisFunctionId:func-f3e7d7868f44448fb532935768ea2ca1 PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-2182fa1a049f4c1c9eeb641f5292f09f:map[] func-f3e7d7868f44448fb532935768ea2ca1:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]}

In KisFunctionC, row = This is Data1 from Test
In KisFunctionC, row = This is Data2 from Test
In KisFunctionC, row = This is Data3 from Test
context.Background
 ====> After commitCurData, flow_name = flowName1, flow_id = flow-8b607ae6d55048408dae1f4e8f6dca6f
All Level Data =
 map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-f3e7d7868f44448fb532935768ea2ca1:[Data From KisFunctionC, index  0 Data From KisFunctionC, index  1 Data From KisFunctionC, index  2]]

KisFunctionE, flow = &{Id:flow-8b607ae6d55048408dae1f4e8f6dca6f Name:flowName1 Conf:0xc00015a780 Funcs:map[func-2182fa1a049f4c1c9eeb641f5292f09f:0xc0001381e0 func-f3e7d7868f44448fb532935768ea2ca1:0xc000138190] FlowHead:0xc000138190 FlowTail:0xc0001381e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0001381e0 ThisFunctionId:func-2182fa1a049f4c1c9eeb641f5292f09f PrevFunctionId:func-f3e7d7868f44448fb532935768ea2ca1 funcParams:map[func-2182fa1a049f4c1c9eeb641f5292f09f:map[] func-f3e7d7868f44448fb532935768ea2ca1:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-f3e7d7868f44448fb532935768ea2ca1:[Data From KisFunctionC, index  0 Data From KisFunctionC, index  1 Data From KisFunctionC, index  2]] inPut:[Data From KisFunctionC, index  0 Data From KisFunctionC, index  1 Data From KisFunctionC, index  2]}

In KisFunctionE, row = Data From KisFunctionC, index  0
In KisFunctionE, row = Data From KisFunctionC, index  1
In KisFunctionE, row = Data From KisFunctionC, index  2
--- PASS: TestNewKisFlowData (0.00s)
PASS
ok      kis-flow/test   0.636s

经过日志的详细校验，结果是符合我们预期的。

好了，目前数据流的最简单版本已经实现了，下一章我们将Function的业务逻辑开放给开发者，而不是写在KisFlow框架中.

3.5 【V0.2】源代码

https://github.com/aceld/kis-flow/releases/tag/v0.2

---

作者:刘丹冰Aceld github: https://github.com/aceld

KisFlow开源项目地址：https://github.com/aceld/kis-flow

gokis-flow golang 流式计算

0 人点赞