ケース (III) - KisFlow-Golang Stream Real- マルチゴルーチンでの KisFlow のアプリケーション

Github: https://github.com/aceld/kis-flow
ドキュメント: https://github.com/aceld/kis-flow/wiki

---

パート 1-概要
Part2.1-プロジェクト構築/基本モジュール
Part2.2-プロジェクト構築/基本モジュール
Part3-データストリーム
Part4 - 機能のスケジューリング
Part5-コネクタ
Part6 - 設定のインポートとエクスポート
Part7 - KisFlow アクション
Part8 - キャッシュ/パラメータ データのキャッシュとデータ パラメータ
Part9 - フローの複数のコピー
Part10-Prometheus メトリクス統計
Part11 - リフレクションに基づく FaaS パラメーター タイプの適応的登録

---

ケース 1 - クイックスタート
Case2-Flow並列運転
Case3 - マルチゴルーチンでの KisFlow の適用
Case4-メッセージキュー (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.Fork() 関数を使用して、分離されたメモリを持つ同じ構成のフロー インスタンスのクローンを作成できます。その後、各フロー インスタンスを異なるゴルーチンで実行して、それぞれのデータ ストリームを計算できます。

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 ケース (III) - KisFlow-Golang Stream Real- マルチゴルーチンでの KisFlow のアプリケーション 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)
}

このコード スニペットでは、Flow1 とそのクローン (FlowClone1) を同時に実行する 2 つのゴルーチンを開始し、生徒 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-概要
Part2.1-プロジェクト構築/基本モジュール
Part2.2-プロジェクト構築/基本モジュール
Part3-データストリーム
Part4 - 機能のスケジューリング
Part5-コネクタ
Part6 - 設定のインポートとエクスポート
Part7 - KisFlow アクション
Part8 - キャッシュ/パラメータ データのキャッシュとデータ パラメータ
Part9 - フローの複数のコピー
Part10-Prometheus メトリクス統計
Part11 - リフレクションに基づく FaaS パラメーター タイプの適応的登録

---

ケース 1 - クイックスタート
Case2-Flow並列運転
Case3 - マルチゴルーチンでの KisFlow の適用
Case4-メッセージキュー (MQ) アプリケーションの KisFlow

以上がケース (III) - KisFlow-Golang Stream Real- マルチゴルーチンでの KisFlow のアプリケーションの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。