feat: bizflow框架开发

Author: logikoisto

common/bizflow/README.md

@@ -0,0 +1,3 @@
+# 一个基于DAG的业务流程框架
+* 详情参考:https://hardcore-tech.feishu.cn/wiki/Ji9BwqCOUiSxXckiixTcvKCKnXb
+* 该框架的工作负载大概在千万量级，后续会在第三期做大幅度的性能优化，使其可在亿级流量场景可用。

common/bizflow/bizflow_test.go

@@ -0,0 +1,105 @@
+package bizflow
+
+import (
+	"context"
+	"testing"
+)
+
+type testFlow struct {
+}
+
+func (f *testFlow) Name() FlowName {
+	return "test_flow"
+}
+func (f *testFlow) BuildGraph(e *Engine) *Graph {
+	g := e.InitGraph()
+	tt := &testNode{}
+	ttB := &testNodeB{}
+	ttC := &testNodeC{}
+	ttD := &testNodeD{}
+	g.AddNode(tt).AddNode(ttB).AddNode(ttC).AddNode(ttD)
+	return g
+}
+
+type testNode struct {
+	in string
+}
+
+func (t *testNode) Name() NodeName {
+	return "testA"
+}
+func (t *testNode) Deps() []NodeName {
+	return nil
+}
+func (t *testNode) Meta() *Meta {
+	return &Meta{}
+}
+func (t *testNode) Run(f *Graph) error {
+	return nil
+}
+
+type testNodeB struct {
+}
+
+func (t *testNodeB) Name() NodeName {
+	return "testB"
+}
+func (t *testNodeB) Deps() []NodeName {
+	return []NodeName{"testA"}
+}
+func (t *testNodeB) Meta() *Meta {
+	return &Meta{}
+}
+func (t *testNodeB) Run(f *Graph) error {
+	return nil
+}
+
+type testNodeC struct {
+}
+
+func (t *testNodeC) Name() NodeName {
+	return "testC"
+}
+func (t *testNodeC) Deps() []NodeName {
+	return []NodeName{"testA"}
+}
+func (t *testNodeC) Meta() *Meta {
+	return &Meta{}
+}
+func (t *testNodeC) Run(f *Graph) error {
+	return nil
+}
+
+type testNodeD struct {
+	out string
+}
+
+func (t *testNodeD) Name() NodeName {
+	return "testD"
+}
+func (t *testNodeD) Deps() []NodeName {
+	return []NodeName{"testB", "testC"}
+}
+func (t *testNodeD) Meta() *Meta {
+	return &Meta{}
+}
+func (t *testNodeD) Run(g *Graph) error {
+	if node := g.GetNode("testA"); node != nil {
+		testA, _ := node.(*testNode)
+		t.out = testA.in
+	}
+	return nil
+}
+func TestBizflow(t *testing.T) {
+	// 进程初始化
+	eng := NewEngine(64)
+	eng.AddFlow(&testFlow{})
+
+	// 请求到来时
+	g := eng.CreateDAG("test_flow")
+	testA := g.Input().(*testNode)
+	testA.in = "hello bizflow"
+	g.Run(context.TODO())
+	testD := g.Output().(*testNodeD)
+	t.Logf("TestBizflow %+v", testD.out)
+}

common/bizflow/engine.go

@@ -0,0 +1,51 @@
+package bizflow
+
+import (
+	"github.com/panjf2000/ants"
+)
+
+type Engine struct {
+	flows           map[FlowName]Flow
+	flowParallelNum int
+	workPool        *ants.Pool
+}
+
+func NewEngine(workSize int) *Engine {
+	e := &Engine{flows: map[FlowName]Flow{}}
+	if wPool, err := ants.NewPool(workSize); err != nil {
+		panic(err)
+	} else {
+		e.workPool = wPool
+	}
+	return e
+}
+
+func (e *Engine) AddFlow(f Flow) *Engine {
+	e.flows[f.Name()] = f
+	// 这里构建一次图是为了在进程初始化时就检查DAG是否合法
+	f.BuildGraph(e)
+	return e
+}
+
+func (e *Engine) FlowParallelNum(num int) *Engine {
+	e.flowParallelNum = num
+	return e
+}
+func (e *Engine) CreateDAG(name FlowName) *Graph {
+	if f := e.getFlow(name); f != nil {
+		g := f.BuildGraph(e)
+		return g
+	}
+	return nil
+}
+
+func (e *Engine) InitGraph() *Graph {
+	return &Graph{e: e, nodes: map[NodeName]*item{}, eventChan: make(chan *item, e.flowParallelNum)}
+}
+
+func (e *Engine) getFlow(name FlowName) Flow {
+	if flow, ok := e.flows[name]; ok {
+		return flow
+	}
+	return nil
+}

common/bizflow/flow.go

@@ -0,0 +1,7 @@
+package bizflow
+
+type FlowName string
+type Flow interface {
+	Name() FlowName
+	BuildGraph(e *Engine) *Graph
+}

common/bizflow/graph.go

@@ -0,0 +1,189 @@
+package bizflow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/bytedance/gopkg/util/logger"
+)
+
+type Graph struct {
+	nodes         map[NodeName]*item
+	lastNodeName  NodeName
+	firstNodeName NodeName
+	eventChan     chan *item
+	e             *Engine
+}
+
+func (g *Graph) AddNode(node Node) *Graph {
+	it := &item{node: node, nexts: map[NodeName]*item{}}
+	nodeName := node.Name()
+	deps := node.Deps()
+	meta := node.Meta()
+	g.nodes[nodeName] = it
+	if len(deps) == 0 {
+		if len(g.firstNodeName) == 0 {
+			g.firstNodeName = nodeName
+		} else {
+			panic("There cannot be any independent stages except for the first node.")
+		}
+	}
+	g.lastNodeName = nodeName
+	it.depsNum = len(deps)
+	it.retryNum = meta.RetryNum
+	for _, parent := range deps {
+		if parentItem := g.getItem(parent); parentItem != nil {
+			if parent == nodeName {
+				panic("The node has a dependency on itself.")
+			}
+			parentItem.nexts[nodeName] = it
+		} else {
+			panic(fmt.Sprintf("The dependent %s is not initialized. There is a cyclic dependency:[%s<-%s] "+
+				"Please ensure that the order of adding nodes during buildgraph follows the precedence order.", parent, nodeName, parent))
+		}
+	}
+	return g
+}
+
+func (g *Graph) GetNode(name NodeName) Node {
+	if it := g.getItem(name); it != nil {
+		event := it.getEvent()
+		if event == execOK {
+			return it.node
+		} else {
+			panic("The node you obtained has not been completed yet. Please follow the declared DAG dependency rules.")
+		}
+	}
+	return nil
+}
+
+func (g *Graph) Input() Node {
+	return g.getItem(g.firstNodeName).node
+}
+
+func (g *Graph) Output() Node {
+	return g.GetNode(g.lastNodeName)
+}
+
+func (g *Graph) getItem(name NodeName) *item {
+	if it, ok := g.nodes[name]; ok {
+		return it
+	}
+	return nil
+}
+
+func (g *Graph) Run(ctx context.Context) error {
+	if len(g.firstNodeName) == 0 {
+		return errors.New("Graph has not been created")
+	}
+	firstIt := g.getItem(g.firstNodeName)
+	if g.firstNodeName == g.lastNodeName {
+		return firstIt.node.Run(g)
+	}
+	g.e.workPool.Submit(func() { g.work(firstIt) })
+	for it := range g.eventChan {
+		switch it.event {
+		case execOK:
+			g.execNext(it)
+			g.tryStop(it)
+		case RetryableError:
+			if it.retryNum > 0 {
+				g.e.workPool.Submit(func() { g.work(it) })
+			}
+		case NonRetryable:
+			logger.CtxErrorf(ctx, "bizflow.masterRun NonRetryable.err=%+v", it.err)
+			g.tryStop(it)
+			return it.err
+		case AbortErr:
+			close(g.eventChan)
+			return nil
+		}
+	}
+	return nil
+}
+
+func (g *Graph) tryStop(it *item) {
+	if g.lastNodeName == it.node.Name() {
+		close(g.eventChan)
+	}
+}
+
+func (g *Graph) execNext(it *item) {
+	for _, itt := range it.nexts {
+		itt.depsNum -= 1
+		if itt.depsNum <= 0 {
+			g.e.workPool.Submit(func() { g.work(itt) })
+		}
+	}
+}
+func (g *Graph) masterChanClosed() bool {
+	select {
+	case _, ok := <-g.eventChan:
+		return !ok
+	default:
+		return false
+	}
+}
+func (g *Graph) work(it *item) {
+	// 先检查一下
+	if g.masterChanClosed() {
+		return
+	}
+	defer func() {
+		if !g.masterChanClosed() {
+			g.eventChan <- it
+		}
+	}()
+	err := it.node.Run(g)
+	if err == nil {
+		it.setEvent(execOK)
+		return
+	}
+	it.err = err
+	meta := it.node.Meta()
+	if meta.IsRetryErr[err] {
+		it.setEvent(RetryableError)
+		return
+	}
+	if meta.IsNonRetryErr[err] {
+		it.setEvent(NonRetryable)
+		return
+	}
+	if meta.AbortErr[err] {
+		it.setEvent(AbortErr)
+		return
+	}
+}
+
+type eventName string
+
+const (
+	execOK         eventName = "execOK"
+	RetryableError eventName = "Retryable error"
+	NonRetryable   eventName = "NonRetryable error"
+	AbortErr       eventName = "abort error"
+)
+
+type item struct {
+	node     Node
+	event    eventName
+	nexts    map[NodeName]*item
+	err      error
+	depsNum  int
+	retryNum int
+	sync.RWMutex
+}
+
+func (it *item) getEvent() eventName {
+	it.RLock()
+	defer it.RUnlock()
+	return it.event
+}
+
+func (it *item) setEvent(e eventName) {
+	it.Lock()
+	defer it.Unlock()
+	it.event = e
+}

common/bizflow/meta.go

@@ -0,0 +1,8 @@
+package bizflow
+
+type Meta struct {
+	RetryNum      int
+	IsRetryErr    map[error]bool
+	IsNonRetryErr map[error]bool
+	AbortErr      map[error]bool
+}

common/bizflow/node.go

@@ -0,0 +1,9 @@
+package bizflow
+
+type NodeName string
+type Node interface {
+	Name() NodeName
+	Deps() []NodeName
+	Meta() *Meta
+	Run(g *Graph) error
+}