共计 8906 个字符,预计需要花费 23 分钟才能阅读完成。
连载中 …
Golang 框架实战 -KisFlow 流式计算框架(1)- 概述
Golang 框架实战 -KisFlow 流式计算框架(2)- 我的项目构建 / 根底模块 -(上)
Golang 框架实战 -KisFlow 流式计算框架(3)- 我的项目构建 / 根底模块 -(下)
首先咱们要先定义 KisFlow 的外围构造体,KisFlow
构造体,通过上述的设计理念,咱们得悉,KisFlow 示意整个一条数据计算流的构造。其中每次数据在一个 Flow 上,顺次执行挂载在以后 Flow 的 Function。
2.3.1 KisFunction 家族
KisFunction 应该是一个链式调用,所以构造体的根本状态应该是一个链表,通过一次 Function 的调用完结后,默认能够调度到下一个 Function 的节点上。在 KisFlow 中,一共有 save
,load
, calculate
, extend
, varify
等多种行为的 Funciton,所以这里咱们采纳上述五种 function 的模板类,不便今后在不同针对不同特色的 function 做更加灵便和性能隔离的拓展和革新。
整体的 KisFunction 的类图设计如下:
2.2.2 形象层 KisFunction 定义
在 kis-flow
中创立一个新的目录 function
用来寄存 function 的代码。
首先形象接口编写在 kis/
目录下。
kis-flow/kis/function.go
package kis
import (
"context"
"kis-flow/config"
)
// Function 流式计算根底计算模块,KisFunction 是一条流式计算的根本计算逻辑单元,// 任意个 KisFunction 能够组合成一个 KisFlow
type Function interface {
// Call 执行流式计算逻辑
Call(ctx context.Context, flow Flow) error
// SetConfig 给以后 Function 实例配置策略
SetConfig(s *config.KisFuncConfig) error
// GetConfig 获取以后 Function 实例配置策略
GetConfig() *config.KisFuncConfig
// SetFlow 给以后 Function 实例设置所依赖的 Flow 实例
SetFlow(f Flow) error
// GetFlow 获取以后 Functioin 实力所依赖的 Flow
GetFlow() Flow
// CreateId 给以后 Funciton 实力生成一个随机的实例 KisID
CreateId()
// GetId 获取以后 Function 的 FID
GetId() string
// GetPrevId 获取以后 Function 上一个 Function 节点 FID
GetPrevId() string
// GetNextId 获取以后 Function 下一个 Function 节点 FID
GetNextId() string
// Next 返回下一层计算流 Function,如果以后层为最初一层,则返回 nil
Next() Function
// Prev 返回上一层计算流 Function,如果以后层为最初一层,则返回 nil
Prev() Function
// SetN 设置下一层 Function 实例
SetN(f Function)
// SetP 设置上一层 Function 实例
SetP(f Function)
}
2.2.3 KisId 随机惟一实例 ID
上述提出了一个新的概念 KisId
。KisID 为Function
的实例 ID,用于 KisFlow 外部辨别不同的实例对象。KisId 和 Function Config 中的 Fid 的区别在于,Fid 用来形容一类 Funcion 策略的 ID,而 KisId 则为在 KisFlow 中 KisFunction 曾经实例化的 实例对象 ID 这个 ID 是随机生成且惟一。
创立 kis-flow/id/
目录,且创立 kis_id.go 文件,实现无关 kis_id 生成的算法。
kis-flow/id/kis_id.go
package id
import (
"github.com/google/uuid"
"kis-flow/common"
"strings"
)
// KisID 获取一个中随机实例 ID
// 格局为 "prefix1-[prefix2-][prefix3-]ID"
// 如:flow-1234567890
// 如:func-1234567890
// 如: conn-1234567890
// 如: func-1-1234567890
func KisID(prefix ...string) (kisId string) {idStr := strings.Replace(uuid.New().String(), "-", "", -1)
kisId = formatKisID(idStr, prefix...)
return
}
func formatKisID(idStr string, prefix ...string) string {
var kisId string
for _, fix := range prefix {
kisId += fix
kisId += common.KisIdJoinChar
}
kisId += idStr
return kisId
}
kisId
模块提供 KisID()
办法,这外面依赖了第三方分布式 ID 生成库 github.com/google/uuid
,生成的随机 ID 为一个字符串,且调用者能够提供多个前缀,通过-
符号进行拼接,失去的随机字符串 ID,如:func-1234567890
针对 KisId 的前缀,提供了一些字符串的枚举,如下:
kis-flow/common/const.go
// KisIdType 用户生成 KisId 的字符串前缀
const (
KisIdTypeFlow = "flow"
KisIdTypeConnnector = "conn"
KisIdTypeFunction = "func"
KisIdTypeGlobal = "global"
KisIdJoinChar = "-"
)
2.2.4 BaseFunction 根底父类
依照设计,咱们须要提供一个 BaseFunction
作为 Function
的一个子类,实现一些根底的性能接口。留下 Call()
让具体模式的 KisFunctionX
来重写实现,上面来进行对 BaseFunction 构造的定义。
在 kis-flow/function/
创立kis_base_funciton.go
文件。
A. 构造定义
kis-flow/function/kis_base_function.go
package function
import (
"context"
"errors"
"kis-flow/common"
"kis-flow/config"
"kis-flow/id"
"kis-flow/kis"
)
type BaseFunction struct {
// Id , KisFunction 的实例 ID,用于 KisFlow 外部辨别不同的实例对象
Id string
Config *config.KisFuncConfig
// flow
Flow kis.Flow // 上下文环境 KisFlow
// link
N kis.Function // 下一个流计算 Function
P kis.Function // 上一个流计算 Function
}
B. 办法实现
kis-flow/function/kis_base_function.go
// Call
// BaseFunction 为空实现,目标为了让其余具体类型的 KisFunction,如 KisFunction_V 来继承 BaseFuncion 来重写此办法
func (base *BaseFunction) Call(ctx context.Context, flow kis.Flow) error {return nil}
func (base *BaseFunction) Next() kis.Function {return base.N}
func (base *BaseFunction) Prev() kis.Function {return base.P}
func (base *BaseFunction) SetN(f kis.Function) {base.N = f}
func (base *BaseFunction) SetP(f kis.Function) {base.P = f}
func (base *BaseFunction) SetConfig(s *config.KisFuncConfig) error {
if s == nil {return errors.New("KisFuncConfig is nil")
}
base.Config = s
return nil
}
func (base *BaseFunction) GetId() string {return base.Id}
func (base *BaseFunction) GetPrevId() string {
if base.P == nil {
//Function 为首结点
return common.FunctionIdFirstVirtual
}
return base.P.GetId()}
func (base *BaseFunction) GetNextId() string {
if base.N == nil {
//Function 为尾结点
return common.FunctionIdLastVirtual
}
return base.N.GetId()}
func (base *BaseFunction) GetConfig() *config.KisFuncConfig {return base.Config}
func (base *BaseFunction) SetFlow(f kis.Flow) error {
if f == nil {return errors.New("KisFlow is nil")
}
base.Flow = f
return nil
}
func (base *BaseFunction) GetFlow() kis.Flow {return base.Flow}
func (base *BaseFunction) CreateId() {base.Id = id.KisID(common.KisIdTypeFunction)
}
这里留神 GetPrevId()
和 GetNextId()
两个办法实现,因为如果以后 Functioin
为双向链表中的第一个节点或者最初一个节点,那么他们的上一个或者下一个是没有节点的,那么 ID 也就不存在,为了在应用中不呈现得不到 ID 的状况,咱们提供了两个虚构 FID,做非凡状况的边界解决,定义在 const.go 中。
kis-flow/common/const.go
const (
// FunctionIdFirstVirtual 为首结点 Function 上一层虚构的 Function ID
FunctionIdFirstVirtual = "FunctionIdFirstVirtual"
// FunctionIdLastVirtual 为尾结点 Function 下一层虚构的 Function ID
FunctionIdLastVirtual = "FunctionIdLastVirtual"
)
2.2.5 KisFunction 的 V /S/L/C/ E 等模式类定义
上面别离实现 V /S/L/C/E 五种不同模式的 KisFunction 子类。这里别离用创立文件来实现。
A. KisFunctionV
kis-flow/function/kis_function_v.go
package function
import (
"context"
"fmt"
"kis-flow/kis"
)
type KisFunctionV struct {BaseFunction}
func (f *KisFunctionV) Call(ctx context.Context, flow kis.Flow) error {fmt.Printf("KisFunctionV, flow = %+v\n", flow)
// TODO 调用具体的 Function 执行办法
return nil
}
B. KisFunctionS
kis-flow/function/kis_function_s.go
package function
import (
"context"
"fmt"
"kis-flow/kis"
)
type KisFunctionS struct {BaseFunction}
func (f *KisFunctionS) Call(ctx context.Context, flow kis.Flow) error {fmt.Printf("KisFunctionS, flow = %+v\n", flow)
// TODO 调用具体的 Function 执行办法
return nil
}
C. KisFunctionL
kis-flow/function/kis_function_l.go
package function
import (
"context"
"fmt"
"kis-flow/kis"
)
type KisFunctionL struct {BaseFunction}
func (f *KisFunctionL) Call(ctx context.Context, flow kis.Flow) error {fmt.Printf("KisFunctionL, flow = %+v\n", flow)
// TODO 调用具体的 Function 执行办法
return nil
}
D. KisFunctionC
kis-flow/function/kis_function_c.go
package function
import (
"context"
"fmt"
"kis-flow/kis"
)
type KisFunctionC struct {BaseFunction}
func (f *KisFunctionC) Call(ctx context.Context, flow kis.Flow) error {fmt.Printf("KisFunction_C, flow = %+v\n", flow)
// TODO 调用具体的 Function 执行办法
return nil
}
E. KisFunctionE
kis-flow/function/kis_function_e.go
package function
import (
"context"
"fmt"
"kis-flow/kis"
)
type KisFunctionE struct {BaseFunction}
func (f *KisFunctionE) Call(ctx context.Context, flow kis.Flow) error {fmt.Printf("KisFunctionE, flow = %+v\n", flow)
// TODO 调用具体的 Function 执行办法
return nil
}
2.2.6 创立 KisFunction 实例
上面提供一个创立具体模式 Function
的办法,这里采纳简略工厂办法模式来实现创建对象。
kis-flow/function/kis_base_function.go
func (base *BaseFunction) CreateId() {base.Id = id.KisID(common.KisIdTypeFunction)
}
// NewKisFunction 创立一个 NsFunction
// flow: 以后所属的 flow 实例
// s : 以后 function 的配置策略
func NewKisFunction(flow kis.Flow, config *config.KisFuncConfig) kis.Function {
var f kis.Function
// 工厂生产泛化对象
switch common.KisMode(config.FMode) {
case common.V:
f = new(KisFunctionV)
break
case common.S:
f = new(KisFunctionS)
case common.L:
f = new(KisFunctionL)
case common.C:
f = new(KisFunctionC)
case common.E:
f = new(KisFunctionE)
default:
//LOG ERROR
return nil
}
// 生成随机实例惟一 ID
f.CreateId()
// 设置根底信息属性
if err := f.SetConfig(config); err != nil {panic(err)
}
if err := f.SetFlow(flow); err != nil {panic(err)
}
return f
}
留神 NewKisFunction()
办法返回的是一个形象的 interface Function
。
还要留神,目前到这里没有实现 Flow
对象,然而 KisFunciton 的创立须要依赖传递一个 Flow
对象,咱们当初能够临时简略创立一个 Flow
对象的构造方法,之后在实现 Flow
章节再欠缺这部分的代码。
在 kis-filw/kis/
中创立 flow.go
文件。
kis-flow/kis/flow.go
package kis
import (
"context"
"kis-flow/config"
)
type Flow interface {// TODO}
在 kis-flow/flow/
下创立 kis_flow.go
文件,实现如下:
kis-flow/flow/kis_flow.go
package flow
import "kis-flow/config"
// KisFlow 用于贯通整条流式计算的上下文环境
type KisFlow struct {
Id string
Name string
// TODO
}
// TODO for test
// NewKisFlow 创立一个 KisFlow.
func NewKisFlow(conf *config.KisFlowConfig) kis.Flow {flow := new(KisFlow)
// 根底信息
flow.Id = id.KisID(common.KisIdTypeFlow)
flow.Name = conf.FlowName
return flow
}
2.2.7 单元测试 KisFunction 创立实例
当初来对上述的 KisFunction 实例的创立做一个简略的单元测试,在 kis-flow/test/
创立 kis_function_test.go
文件。
kis-flow/test/kis_function_test.go
package test
import (
"context"
"kis-flow/common"
"kis-flow/config"
"kis-flow/flow"
"kis-flow/function"
"testing"
)
func TestNewKisFunction(t *testing.T) {ctx := context.Background()
// 1. 创立一个 KisFunction 配置实例
source := config.KisSource{
Name: "公众号抖音商城户订单数据",
Must: []string{"order_id", "user_id"},
}
myFuncConfig1 := config.NewFuncConfig("funcName1", common.C, &source, nil)
if myFuncConfig1 == nil {panic("myFuncConfig1 is nil")
}
// 2. 创立一个 KisFlow 配置实例
myFlowConfig1 := config.NewFlowConfig("flowName1", common.FlowEnable)
// 3. 创立一个 KisFlow 对象
flow1 := flow.NewKisFlow(myFlowConfig1)
// 4. 创立一个 KisFunction 对象
func1 := function.NewKisFunction(flow1, myFuncConfig1)
if err := func1.Call(ctx, flow1); err != nil {t.Errorf("func1.Call() error = %v", err)
}
}
流程很简略,分为四个小步骤:
- 创立一个 KisFunction 配置实例
- 创立一个 KisFlow 配置实例
- 创立一个 KisFlow 对象
- 创立一个 KisFunction 对象
cd 到 kis-flow/test/
目录下执行:
go test -test.v -test.paniconexit0 -test.run TestNewKisFunction
后果如下:
=== RUN TestNewKisFunction
KisFunctionC, flow = &{Id:flow-866de5abc8134fc9bb8e5248a3ce7137 Name:flowName1 Conf:0xc00014e780 Funcs:map[] FlowHead:<nil> FlowTail:<nil> flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:<nil> ThisFunctionId: PrevFunctionId: funcParams:map[] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0}}
--- PASS: TestNewKisFunction (0.00s)
PASS
ok kis-flow/test 1.005s
咱们曾经调用到了具体的 KisFunciton_C
实例的 Call()
办法。
作者: 刘丹冰 Aceld github: https://github.com/aceld
KisFlow 开源我的项目地址:https://github.com/aceld/kis-flow
连载中 …
Golang 框架实战 -KisFlow 流式计算框架(1)- 概述
Golang 框架实战 -KisFlow 流式计算框架(2)- 我的项目构建 / 根底模块 -(上)
Golang 框架实战 -KisFlow 流式计算框架(3)- 我的项目构建 / 根底模块 -(下)