案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用-Golang-PHP中文網

案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用

WBOY

發布： 2024-07-16 07:23:18

原創

533 人瀏覽過

Case (III) - KisFlow-Golang Stream Real- Application of KisFlow in Multi-案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用

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 開發者文件

原始碼範例

https://github.com/aceld/kis-flow-usage/tree/main/6-flow_in_goroutines

如果需要同一個Flow在多個Goroutine中並發運行，可以使用flow.Fork()函數克隆一個記憶體隔離但配置相同的Flow實例。然後，每個 Flow 實例可以在不同的 Goroutine 中執行，以計算各自的資料流。

案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用

package main

import (
    "context"
    "fmt"
    "github.com/aceld/kis-flow/file"
    "github.com/aceld/kis-flow/kis"
    "sync"
)

func main() {
    ctx := context.Background()
    // Get a WaitGroup
    var wg sync.WaitGroup

    // Load Configuration from file
    if err := file.ConfigImportYaml("conf/"); err != nil {
        panic(err)
    }

    // Get the flow
    flow1 := kis.Pool().GetFlow("CalStuAvgScore")
    if flow1 == nil {
        panic("flow1 is nil")
    }
    // Fork the flow
    flowClone1 := flow1.Fork(ctx)

    // Add to WaitGroup
    wg.Add(2)

    // Run Flow1
    go func() {
        defer wg.Done()
        // Submit a string
        _ = flow1.CommitRow(`{"stu_id":101, "score_1":100, "score_2":90, "score_3":80}`)
        // Submit a string
        _ = flow1.CommitRow(`{"stu_id":1001, "score_1":100, "score_2":70, "score_3":60}`)

        // Run the flow
        if err := flow1.Run(ctx); err != nil {
            fmt.Println("err: ", err)
        }
    }()

    // Run FlowClone1
    go func() {
        defer wg.Done()
        // Submit a string
        _ = flowClone1.CommitRow(`{"stu_id":201, "score_1":100, "score_2":90, "score_3":80}`)
        // Submit a string
        _ = flowClone1.CommitRow(`{"stu_id":2001, "score_1":100, "score_2":70, "score_3":60}`)

        if err := flowClone1.Run(ctx); err != nil {
            fmt.Println("err: ", err)
        }
    }()

    // Wait for 案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用 to finish
    wg.Wait()

    fmt.Println("All flows completed.")

    return
}

func init() {
    // Register functions
    kis.Pool().FaaS("VerifyStu", VerifyStu)
    kis.Pool().FaaS("AvgStuScore", AvgStuScore)
    kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
}

登入後複製

在此程式碼片段中，我們啟動兩個案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用來同時執行 Flow1 及其複製 (FlowClone1)，以計算學生 101、1001、201 和 2001 的最終平均分數。

作者：Aceld
GitHub：https://github.com/aceld

KisFlow開源專案位址：https://github.com/aceld/kis-flow

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

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

以上是案例（三）-KisFlow-Golang Stream實戰-KisFlow在多Goroutine中的應用的詳細內容。更多資訊請關注PHP中文網其他相關文章！