Go语言——goflow工作流使用

一、引入依赖

这个很坑，他不答应连接带暗码的redis，只能使用不带暗码的redis，要带暗码的话得自己改一下源代码，无语

1. go get github.com/s8sg/goflow

复制代码

二、画出我们的工作流程

三、编写代码

1. package main
3. import (
4.         "encoding/json"
5.         "fmt"
6.         flow "github.com/s8sg/goflow/flow/v1"
7.         goflow "github.com/s8sg/goflow/v1"
8.         "log"
9.         "math/rand"
10.         "strconv"
11. )
13. // Input 输入一个数字
14. func Input(data []byte, option map[string][]string) ([]byte, error) {
15.         var input map[string]int
16.         // 获取输入的数
17.         if err := json.Unmarshal(data, &input); err != nil {
18.                 return nil, err
19.         }
20.         outputInt := input["input"]
21.         // 将数据交给工作流处理
22.         return []byte(strconv.Itoa(outputInt)), nil
23. }
25. // AddOne 加上10以内的一个随机整数
26. func AddOne(data []byte, option map[string][]string) ([]byte, error) {
27.         // 获取上一个工作流的数据
28.         num, _ := strconv.Atoi(string(data))
29.         outputInt := num + rand.Intn(10) + 1
30.         fmt.Println("AddOne = ", outputInt)
31.         // 交给下一个工作流处理
32.         return []byte(strconv.Itoa(outputInt)), nil
33. }
35. // AddTwo 加上10以内的一个随机整数
36. func AddTwo(data []byte, option map[string][]string) ([]byte, error) {
37.         num, _ := strconv.Atoi(string(data))
38.         outputInt := num + rand.Intn(10) + 1
39.         fmt.Println("AddTwo = ", outputInt)
40.         return []byte(strconv.Itoa(outputInt)), nil
41. }
43. // Aggregator 聚合节点
44. func Aggregator(data []byte, option map[string][]string) ([]byte, error) {
45.         fmt.Println("Aggregator = ", string(data))
46.         return data, nil
47. }
49. // Expand10 扩大10倍
50. func Expand10(data []byte, option map[string][]string) ([]byte, error) {
51.         num, _ := strconv.Atoi(string(data))
52.         outputInt := num * 10
53.         fmt.Println("Expand10 = ", outputInt)
54.         return []byte(strconv.Itoa(outputInt)), nil
55. }
57. // Expand100 扩大100倍
58. func Expand100(data []byte, option map[string][]string) ([]byte, error) {
59.         num, _ := strconv.Atoi(string(data))
60.         outputInt := num * 100
61.         fmt.Println("Expand100 = ", outputInt)
62.         return []byte(strconv.Itoa(outputInt)), nil
63. }
65. // Output 输出节点
66. func Output(data []byte, option map[string][]string) ([]byte, error) {
67.         fmt.Println("Output = ", string(data))
68.         return data, nil
69. }
71. // 定义我们自己的一个流程
72. func MyFlow(workflow *flow.Workflow, context *flow.Context) error {
73.         // 创建DAG
74.         dag := workflow.Dag()
75.         // 创建节点
76.         dag.Node("input", Input)
77.         dag.Node("add-one", AddOne)
78.         dag.Node("add-two", AddTwo)
79.         // 这个聚合节点，就需要拿到add-one和add-two的结果
80.         dag.Node("aggregator", Aggregator, flow.Aggregator(func(m map[string][]byte) ([]byte, error) {
81.                 addOneResult, _ := strconv.Atoi(string(m["add-one"]))
82.                 addTwoResult, _ := strconv.Atoi(string(m["add-two"]))
83.                 num := addOneResult + addTwoResult
84.                 fmt.Println("aggregator = ", num)
85.                 return []byte(strconv.Itoa(num)), nil
86.         }))
87.         // 这个方式是获取到节点的数据进行判断，然后返回一个字符串数组
88.         f1 := func(bytes []byte) []string {
89.                 num, _ := strconv.Atoi(string(bytes))
90.                 fmt.Println("ConditionalBranch = ", num)
91.                 if num > 10 {
92.                         return []string{"moreThan"}
93.                 }
94.                 return []string{"lessThan"}
95.         }
96.         // 这个方法就是将分支的数据返回给output
97.         f2 := func(m map[string][]byte) ([]byte, error) {
98.                 if v, ok := m["moreThan"]; ok {
99.                         i, _ := strconv.Atoi(string(v))
100.                         fmt.Println("f2 moreThan = ", i)
101.                         return v, nil
102.                 }
103.                 if v, ok := m["lessThan"]; ok {
104.                         i, _ := strconv.Atoi(string(v))
105.                         fmt.Println("f2 lessThan = ", i)
106.                         return v, nil
107.                 }
108.                 return nil, nil
109.         }
110.         // 创建一个条件分支节点
111.         branches := dag.ConditionalBranch("judge", []string{"moreThan", "lessThan"}, f1, flow.Aggregator(f2))
112.         branches["moreThan"].Node("expand-10", Expand10)
113.         branches["lessThan"].Node("expand-100", Expand100)
114.         dag.Node("output", Output)
115.         // 构建关系
116.         dag.Edge("input", "add-one")
117.         dag.Edge("input", "add-two")
118.         dag.Edge("add-one", "aggregator")
119.         dag.Edge("add-two", "aggregator")
120.         dag.Edge("aggregator", "judge")
121.         dag.Edge("judge", "output")
122.         return nil
123. }
125. func main() {
126.         fs := goflow.FlowService{
127.                 Port:              10001,
128.                 RedisURL:          "127.0.0.1:6379",
129.                 RedisPwd:          "p@ssw0rd",
130.                 WorkerConcurrency: 5,
131.                 RetryCount:        0,
132.         }
133.         if err := fs.Register("myFlow", MyFlow); err != nil {
134.                 log.Printf("goflow register err: %v\n", err)
135.                 return
136.         }
137.         if err := fs.Start(); err != nil {
138.                 panic(err)
139.         }
140. }

复制代码

四、Postman测试