案例(四)- KisFlow-Golang Stream Real-訊息佇列(MQ)應用中的 KisFlow

WBOY
發布: 2024-07-18 03:44:47
原創
779 人瀏覽過

Case (IV) - KisFlow-Golang Stream Real- KisFlow in Message Queue (MQ) Applications

Github:https://github.com/aceld/kis-flow
文件:https://github.com/aceld/kis-flow/wiki


第 1 部分-概覽
Part2.1-專案建置/基礎模組
Part2.2-專案建置/基礎模組
第三部分-資料流
Part4-功能調度
第5部-連接器
Part6-配置導入導出
Part7-KisFlow 動作
Part8-Cache/Params 資料快取和資料參數
Part9-流程多份
Part10-Prometheus Metrics 統計量
Part11-基於反射的FaaS參數類型自適應註冊


案例1-快速入門
Case2-Flow並行操作
Case3-KisFlow在多Goroutine中的應用
案例4-訊息佇列(MQ)應用中的KisFlow

下載 KisFlow 原始碼

$go get github.com/aceld/kis-flow
登入後複製

KisFlow 開發者文件

KisFlow 與 Kafka

範例原始碼

https://github.com/aceld/kis-flow-usage/tree/main/12-with_kafka

本例中我們使用github.com/segmentio/kafka-go作為第三方Kafka Client SDK(開發者可以選擇其他Kafka Go工具)。

package main import ( "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 file if err := file.ConfigImportYaml("conf/"); err != nil { panic(err) } // Get the flow flowOrg := kis.Pool().GetFlow("CalStuAvgScore") if flowOrg == nil { panic("flowOrg is nil") } // Create a new Kafka reader reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "SourceStuScore", GroupID: "group1", MinBytes: 10e3, // 10KB MaxBytes: 10e6, // 10MB MaxWait: 500 * time.Millisecond, // Maximum wait time StartOffset: kafka.FirstOffset, }) defer reader.Close() var wg sync.WaitGroup for i := 0; i < 5; i++ { // Use 5 consumers to consume in parallel wg.Add(1) go func() { // Fork a new flow for each consumer flowCopy := flowOrg.Fork(ctx) defer wg.Done() for { // Read a message from Kafka message, 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 flow if err := flowCopy.Run(ctx); err != nil { fmt.Println("err: ", err) return } } }() } wg.Wait() return } func init() { // Register functions kis.Pool().FaaS("VerifyStu", VerifyStu) kis.Pool().FaaS("AvgStuScore", AvgStuScore) kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore) }
登入後複製

KisFlow 與 Nsq

範例原始碼:

https://github.com/aceld/kis-flow-usage/tree/main/13-with_nsq

此 KisFlow 消費者使用 github.com/nsqio/go-nsq 作為第三方 SDK。

package main import ( "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 file if err := file.ConfigImportYaml("conf/"); err != nil { panic(err) } // Get the flow flowOrg := kis.Pool().GetFlow("CalStuAvgScore") if flowOrg == nil { panic("flowOrg is nil") } // Create a new NSQ consumer config := nsq.NewConfig() config.MaxInFlight = 5 consumer, 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 message flowCopy := flowOrg.Fork(ctx) // Commit the message to the flow _ = flowCopy.CommitRow(string(message.Body)) // Run the flow if 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 functions kis.Pool().FaaS("VerifyStu", VerifyStu) kis.Pool().FaaS("AvgStuScore", AvgStuScore) kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore) }
登入後複製

KisFlow 與 RocketMQ

範例原始碼:

https://github.com/aceld/kis-flow-usage/tree/main/14-with_rocketmq

使用 github.com/apache/rocketmq-client-go 作為 RocketMQ 消費者 SDK。

package main import ( "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 file if err := file.ConfigImportYaml("conf/"); err != nil { panic(err) } // Get the flow myFlow := kis.Pool().GetFlow("CalStuAvgScore") if myFlow == nil { panic("myFlow is nil") } // Create a new RocketMQ consumer c, 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 _ = myFlow.CommitRow(string(msg.Body)) } // Run the flow if err := myFlow.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

文件:https://github.com/aceld/kis-flow/wiki


第 1 部分-概覽
Part2.1-專案建置/基礎模組
Part2.2-專案建置/基礎模組
第三部分-資料流
Part4-功能調度
第5部-連接器
Part6-配置導入導出
Part7-KisFlow 動作
Part8-Cache/Params 資料快取和資料參數
Part9-流程多份
Part10-Prometheus Metrics 統計量
Part11-基於反射的FaaS參數類型自適應註冊


案例1-快速入門
Case2-Flow並行操作
Case3-KisFlow在多Goroutine中的應用
案例4-訊息佇列(MQ)應用中的KisFlow

以上是案例(四)- KisFlow-Golang Stream Real-訊息佇列(MQ)應用中的 KisFlow的詳細內容。更多資訊請關注PHP中文網其他相關文章!

來源:dev.to
本網站聲明
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
最新下載
更多>
網站特效
網站源碼
網站素材
前端模板
關於我們 免責聲明 Sitemap
PHP中文網:公益線上PHP培訓,幫助PHP學習者快速成長!