KisFlow-Golang流式实时计算案例(四)-KisFlow在消息队列MQ中的应用

Golang框架实战-KisFlow流式计算框架专栏

Golang框架实战-KisFlow流式计算框架(1)-概述
Golang框架实战-KisFlow流式计算框架(2)-项目构建/基础模块-(上)
Golang框架实战-KisFlow流式计算框架(3)-项目构建/基础模块-(下)
Golang框架实战-KisFlow流式计算框架(4)-数据流
Golang框架实战-KisFlow流式计算框架(5)-Function调度
Golang框架实战-KisFlow流式计算框架(6)-Connector
Golang框架实战-KisFlow流式计算框架(7)-配置导入与导出
Golang框架实战-KisFlow流式计算框架(8)-KisFlow Action
Golang框架实战-KisFlow流式计算框架(9)-Cache/Params 数据缓存与数据参数
Golang框架实战-KisFlow流式计算框架(10)-Flow多副本
Golang框架实战-KisFlow流式计算框架(11)-Prometheus Metrics统计
Golang框架实战-KisFlow流式计算框架(12)-基于反射自适应注册FaaS形参类型

案例：
KisFlow-Golang流式计算案例(一)快速开始QuickStart
KisFlow-Golang流式计算案例(二)-Flow并流操作
KisFlow-Golang流式计算案例(二)-KisFlow在多协程中的应用

DownLoad kis-flow source

$go get github.com/aceld/kis-flow

KisFlow with Kafka

案例源代码
https://github.com/aceld/kis-flow-usage/tree/main/12-with_kafka

这里以github.com/segmentio/kafka-go 作为第三方Kafka Client SDK（开发者也可以选择其他kafka的go开发工具）。

package mainimport ("context""fmt""github.com/aceld/kis-flow/file""github.com/aceld/kis-flow/kis""github.com/segmentio/kafka-go""sync""time"
)func main() {ctx := context.Background()// Load Configuration from fileif err := file.ConfigImportYaml("conf/"); err != nil {panic(err)}// Get the flowflowOrg := kis.Pool().GetFlow("CalStuAvgScore")if flowOrg == nil {panic("flowOrg is nil")}// Create a new Kafka readerreader := kafka.NewReader(kafka.ReaderConfig{Brokers:     []string{"localhost:9092"},Topic:       "SourceStuScore",GroupID:     "group1",MinBytes:    10e3,                   // 10KBMaxBytes:    10e6,                   // 10MBMaxWait:     500 * time.Millisecond, // 最长等待时间StartOffset: kafka.FirstOffset,})defer reader.Close()var wg sync.WaitGroupfor i := 0; i < 5; i++ { // use 5 consumers to consume in parallelwg.Add(1)go func() {// fork a new flow for each consumerflowCopy := flowOrg.Fork(ctx)defer wg.Done()for {// Read a message from Kafkamessage, err := reader.ReadMessage(ctx)if err != nil {fmt.Printf("error reading message: %v\n", err)break}// Commit the message to the flow_ = flowCopy.CommitRow(string(message.Value))// Run the flowif err := flowCopy.Run(ctx); err != nil {fmt.Println("err: ", err)return}}}()}wg.Wait()return
}func init() {// Register functionskis.Pool().FaaS("VerifyStu", VerifyStu)kis.Pool().FaaS("AvgStuScore", AvgStuScore)kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
}

KisFlow with Nsq

案例源代码
https://github.com/aceld/kis-flow-usage/tree/main/13-with_nsq

本KisFlow消费者以github.com/nsqio/go-nsq作为第三方SDK。

package mainimport ("context""fmt""github.com/aceld/kis-flow/file""github.com/aceld/kis-flow/kis""github.com/nsqio/go-nsq"
)func main() {ctx := context.Background()// Load Configuration from fileif err := file.ConfigImportYaml("conf/"); err != nil {panic(err)}// Get the flowflowOrg := kis.Pool().GetFlow("CalStuAvgScore")if flowOrg == nil {panic("flowOrg is nil")}// Create a new NSQ consumerconfig := nsq.NewConfig()config.MaxInFlight = 5consumer, err := nsq.NewConsumer("SourceStuScore", "channel1", config)if err != nil {panic(err)}consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {// fork a new flow for each messageflowCopy := flowOrg.Fork(ctx)// Commit the message to the flow_ = flowCopy.CommitRow(string(message.Body))// Run the flowif err := flowCopy.Run(ctx); err != nil {fmt.Println("err: ", err)return err}return nil}))err = consumer.ConnectToNSQLookupd("localhost:4161")if err != nil {panic(err)}defer consumer.Stop()select {}
}func init() {// Register functionskis.Pool().FaaS("VerifyStu", VerifyStu)kis.Pool().FaaS("AvgStuScore", AvgStuScore)kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
}

KisFlow with RocketMQ

案例源代码
https://github.com/aceld/kis-flow-usage/tree/main/14-with_rocketmq

以 github.com/apache/rocketmq-client-go 作为RocketMQ消费者SDK。

package mainimport ("context""fmt""github.com/aceld/kis-flow/file""github.com/aceld/kis-flow/kis""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/consumer""github.com/apache/rocketmq-client-go/v2/primitive"
)func main() {// Load Configuration from fileif err := file.ConfigImportYaml("conf/"); err != nil {panic(err)}// Get the flowmyFloq := kis.Pool().GetFlow("CalStuAvgScore")if myFloq == nil {panic("myFloq is nil")}// Create a new RocketMQ consumerc, err := rocketmq.NewPushConsumer(consumer.WithGroupName("group1"),consumer.WithNameServer([]string{"localhost:9876"}),)if err != nil {panic(err)}err = c.Subscribe("SourceStuScore", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {for _, msg := range msgs {// Commit the message to the flow_ = myFloq.CommitRow(string(msg.Body))}// Run the flowif err := myFloq.Run(ctx); err != nil {fmt.Println("err: ", err)return consumer.ConsumeRetryLater, err}return consumer.ConsumeSuccess, nil})if err != nil {panic(err)}err = c.Start()if err != nil {panic(err)}defer c.Shutdown()select {}
}

作者:刘丹冰Aceld github: https://github.com/aceld
KisFlow开源项目地址：https://github.com/aceld/kis-flow