简体中文(ZH-CN) English(EN) 繁体中文(ZH-TW) 日本語(JA) 한국어(KO) Melayu(MS) Français(FR) Deutsch(DE)
ホームページ > バックエンド開発 > Golang > 本文

ケース (II) - KisFlow-Golang ストリーム リアルタイム コンピューティング - フロー並列操作

PHPz
リリース： 2024-07-16 17:28:10
オリジナル
718 人が閲覧しました

Case (II) - KisFlow-Golang Stream Real-Time Computing - Flow Parallel Operation

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/8-connector

KisFlow はコネクタ経由で 2 つのフローの組み合わせを実現できます

この紹介では、次の 2 つのフローを組み合わせて、コネクタのインターフェイスと使用法について説明します。

データフロー図

Flow Diagram

事例紹介

学生には 4 つの属性があると仮定します。

Student ID: stu_id
Credit 1: score_1
Credit 2: score_2
Credit 3: score_3
ログイン後にコピー

Flow1: CalStuAvgScore-1-2 を定義して、単位 1 (score_1) と単位 2 (score_2) (avg_score_1_2) の学生の平均スコアを計算します。
Flow2: CalStuAvgScore-3 を定義して、単位 3 (score_3) と単位 1、単位 2、および単位 3 の平均である avg_score_1_2 の学生の平均スコアを計算します。単位 1 と単位 2 の平均は、Flow1 によって提供されます。

フロー1

Flow1 は 4 つの関数で構成されます:

V (関数: VerifyStu) StuId
の有効性を検証します。 C (関数: AvgStuScore12) Credit 1 と Credit 2 の平均スコアを計算します
S (関数: SaveScoreAvg12) avg_score_1_2 を Redis
に保存します E (関数: PrintStuAvgScore) クレジット 1 とクレジット 2 の平均スコアを出力します。

フロー2

Flow2 は 4 つの関数で構成されます:

V (関数: VerifyStu) StuId
の有効性を検証します。 L (関数: LoadScoreAvg12) は、Flow1
によって計算された現在の学生の単位 1 と単位 2 の平均スコア (avg_score_1_2) を読み取ります。 C (関数: AvgStuScore3) Credit 3 の平均スコアと Credit 1 と Credit 2 の平均スコアを計算します
E (関数: PrintStuAvgScore) クレジット 1、クレジット 2、クレジット 3 の平均スコアを出力します。

conf/func/func-AvgStuScore-3.yml

kistype: func
fname: AvgStuScore3
fmode: Calculate
source:
    name: SourceStuScore
    must:
        - stu_id
ログイン後にコピー

conf/func/func-LoadScoreAvg-1-2.yml

kistype: func
fname: LoadScoreAvg12
fmode: Load
source:
    name: SourceStuScore
    must:
        - stu_id
option:
    cname: Score12Cache
ログイン後にコピー

基本データプロトコル

stu_proto.go

package main

type StuScore1_2 struct {
    StuId  int `json:"stu_id"`
    Score1 int `json:"score_1"`
    Score2 int `json:"score_2"`
}

type StuScoreAvg struct {
    StuId    int     `json:"stu_id"`
    AvgScore float64 `json:"avg_score"`
}

type StuScore3 struct {
    StuId      int     `json:"stu_id"`
    AvgScore12 float64 `json:"avg_score_1_2"` // score_1, score_2 avg
    Score3     int     `json:"score_3"`
}
ログイン後にコピー

コネクタの初期化

このプロジェクトで定義されているコネクタ、Score12Cache は、Redis に関連付けられたリンク リソースです。このコネクタには、KisFlow の開始時に接続を確立するための初期化メソッドが必要です。

conn_init.go

package main

import (
    "context"
    "fmt"
    "github.com/aceld/kis-flow/kis"
    "github.com/aceld/kis-flow/log"
    "github.com/go-redis/redis/v8"
)

// type ConnInit func(conn Connector) error

func InitScore12Cache(connector kis.Connector) error {
    fmt.Println("===> Call Connector InitScore12Cache")

    // init Redis Conn Client
    rdb := redis.NewClient(&redis.Options{
        Addr:     connector.GetConfig().AddrString, // Redis-Server address
        Password: "",                               // password
        DB:       0,                                // select db
    })

    // Ping test
    pong, err := rdb.Ping(context.Background()).Result()
    if err != nil {
        log.Logger().ErrorF("Failed to connect to Redis: %v", err)
        return err
    }
    fmt.Println("Connected to Redis:", pong)

    // set rdb to connector
    connector.SetMetaData("rdb", rdb)

    return nil
}
ログイン後にコピー

ここでは、正常に接続された Redis インスタンスがコネクタのキャッシュ変数「rdb」に保存されます。

    // set rdb to connector
    connector.SetMetaData("rdb", rdb)
ログイン後にコピー

FaaSの導入

機能(V): VerifyStu

faas_stu_verify.go

package main

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

type VerifyStuIn struct {
    serialize.DefaultSerialize
    StuId int `json:"stu_id"`
}

func VerifyStu(ctx context.Context, flow kis.Flow, rows []*VerifyStuIn) error {
    fmt.Printf("->Call Func VerifyStu\n")

    for _, stu := range rows {
        // Filter out invalid data
        if stu.StuId < 0 || stu.StuId > 999 {
            // Terminate the current Flow process, subsequent functions of the current Flow will not be executed
            return flow.Next(kis.ActionAbort)
        }
    }

    return flow.Next(kis.ActionDataReuse)
}
ログイン後にコピー

VerifyStu() はデータを検証するために使用されます。データが要件を満たさない場合、現在のデータ フローは終了します。最後に、データは再利用され、flow.Next(kis.ActionDataReuse) を通じて次の層に渡されます。

関数(C): AvgStuScore12

faas_avg_score_1_2.go

package main

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

type AvgStuScoreIn_1_2 struct {
    serialize.DefaultSerialize
    StuScore1_2
}

type AvgStuScoreOut_1_2 struct {
    serialize.DefaultSerialize
    StuScoreAvg
}

func AvgStuScore12(ctx context.Context, flow kis.Flow, rows []*AvgStuScoreIn_1_2) error {
    fmt.Printf("->Call Func AvgStuScore12\n")

    for _, row := range rows {

        out := AvgStuScoreOut_1_2{
            StuScoreAvg: StuScoreAvg{
                StuId:    row.StuId,
                AvgScore: float64(row.Score1+row.Score2) / 2,
            },
        }

        // Submit result data
        _ = flow.CommitRow(out)
    }

    return flow.Next()
}
ログイン後にコピー

AvgStuScore12() は、score_1 とscore_2 の平均スコアを計算し、avg_score となります。

関数: SaveScoreAvg12

faas_save_score_avg_1_2.go

package main

import (
    "context"
    "fmt"
    "github.com/aceld/kis-flow/kis"
    "github.com/aceld/kis-flow/serialize"
    "github.com/go-redis/redis/v8"
    "strconv"
)

type SaveStuScoreIn struct {
    serialize.DefaultSerialize
    StuScoreAvg
}

func BatchSetStuScores(ctx context.Context, conn kis.Connector, rows []*SaveStuScoreIn) error {

    var rdb *redis.Client

    // Get Redis Client
    rdb = conn.GetMetaData("rdb").(*redis.Client)

    // Set data to redis
    pipe := rdb.Pipeline()

    for _, score := range rows {
        // make key
        key := conn.GetConfig().Key + strconv.Itoa(score.StuId)

        pipe.HMSet(context.Background(), key, map[string]interface{}{
            "avg_score": score.AvgScore,
        })
    }

    _, err := pipe.Exec(ctx)
    if err != nil {
        return err
    }

    return nil
}

func SaveScoreAvg12(ctx context.Context, flow kis.Flow, rows []*SaveStuScoreIn) error {
    fmt.Printf("->Call Func SaveScoreAvg12\n")

    conn, err := flow.GetConnector()
    if err != nil {
        fmt.Printf("SaveScoreAvg12(): GetConnector err = %s\n", err.Error())
        return err
    }

    if BatchSetStuScores(ctx, conn, rows) != nil {
        fmt.Printf("SaveScoreAvg12(): BatchSetStuScores err = %s\n", err.Error())
        return err
    }

    return flow.Next(kis.ActionDataReuse)
}
ログイン後にコピー

SaveScoreAvg12() は、コネクタで構成されたキーを使用して、バインドされたコネクタを通じて Redis にデータを保存します。最後に、ソース データは次の関数に透過的に送信されます。

関数(E): PrintStuAvgScore

faas_stu_score_avg_print.go

package main

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

type PrintStuAvgScoreIn struct {
    serialize.DefaultSerialize
    StuId    int     `json:"stu_id"`
    AvgScore float64 `json:"avg_score"`
}

func PrintStuAvgScore(ctx context.Context, flow kis.Flow, rows []*PrintStuAvgScoreIn) error {
    fmt.Printf("->Call Func PrintStuAvgScore, in Flow[%s]\n", flow.GetName())

    for _, row := range rows {
        fmt.Printf("stuid: [%+v], avg score: [%+v]\n", row.StuId, row.AvgScore)
    }

    return flow.Next()
}
ログイン後にコピー

PrintStuAvgScore() は、現在の生徒の平均スコアを出力します。

関数(L): LoadScoreAvg12

faas_load_score_avg_1_2.go

package main

import (
    "context"
    "fmt"
    "github.com/aceld/kis-flow/kis"
    "github.com/aceld/kis-flow/serialize"
    "github.com/go-redis/redis/v8"
    "strconv"
)

type LoadStuScoreIn struct {
    serialize.DefaultSerialize
    StuScore3
}

type LoadStuScoreOut struct {
    serialize.DefaultSerialize
    StuScore3
}

func GetStuScoresByStuId(ctx context.Context, conn kis.Connector, stuId int) (float64, error) {

    var rdb *redis.Client

    // Get Redis Client
    rdb = conn.GetMetaData("rdb").(*redis.Client)

    // make key
    key := conn.GetConfig().Key + strconv.Itoa(stuId)

    // get data from redis
    result, err := rdb.HGetAll(ctx, key).Result()
    if err != nil {
        return 0, err
    }

    // get value
    avgScoreStr, ok := result["avg_score"]
    if !ok {
        return 0, fmt.Errorf("avg_score not found for stuId: %d", stuId)
    }

    // parse to float64
    avgScore, err := strconv.ParseFloat(avgScoreStr, 64)
    if err != nil {
        return 0, err
    }

    return avgScore, nil
}

func LoadScoreAvg12(ctx context.Context, flow kis.Flow, rows []*LoadStuScoreIn) error {
    fmt.Printf("->Call Func LoadScoreAvg12\n")

    conn, err := flow.GetConnector()
    if err != nil {
        fmt.Printf("LoadScoreAvg12(): GetConnector err = %s\n", err.Error())
        return err
    }

    for _, row := range rows {
        stuScoreAvg1_2, err := GetStuScoresByStuId(ctx, conn, row.StuId)
        if err != nil {
            fmt.Printf("LoadScoreAvg12(): GetStuScoresByStuId err = %s\n", err.Error())
            return err
        }

        out := LoadStuScoreOut{
            StuScore3: StuScore3{
                StuId:      row.StuId,
                Score3:     row.Score3,
                AvgScore12: stuScoreAvg1_2, // avg score of score1 and score2 (load from redis)
            },
        }

        // commit result
        _ = flow.CommitRow(out)
    }

    return flow.Next()
}
ログイン後にコピー

LoadScoreAvg12() reads the average score of score_1 and score_2 from Redis through the linked resource Redis of the bound Connector using the key configured in the Connector. It then sends the source data from upstream, along with the newly read average score of score1 and score2, to the next layer.

Function(C): AvgStuScore3

faas_stu_score_avg_3.go

package main

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

type AvgStuScore3In struct {
    serialize.DefaultSerialize
    StuScore3
}

type AvgStuScore3Out struct {
    serialize.DefaultSerialize
    StuScoreAvg
}

func AvgStuScore3(ctx context.Context, flow kis.Flow, rows []*AvgStuScore3In) error {
    fmt.Printf("->Call Func AvgStuScore3\n")

    for _, row := range rows {

        out := AvgStuScore3Out{
            StuScoreAvg: StuScoreAvg{
                StuId:    row.StuId,
                AvgScore: (float64(row.Score3) + row.AvgScore12*2) / 3,
            },
        }

        // Submit result data
        _ = flow.CommitRow(out)
    }

    return flow.Next()
}
ログイン後にコピー

AvgStuScore3() recalculates the average score of three scores by adding score_3 and the average score of score_1 and score_2, resulting in the final average score avg_score.

Register FaaS & CaaSInit/CaaS (Register Function/Connector)

main.go

func init() {
    // Register functions
    kis.Pool().FaaS("VerifyStu", VerifyStu)
    kis.Pool().FaaS("AvgStuScore12", AvgStuScore12)
    kis.Pool().FaaS("SaveScoreAvg12", SaveScoreAvg12)
    kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
    kis.Pool().FaaS("LoadScoreAvg12", LoadScoreAvg12)
    kis.Pool().FaaS("AvgStuScore3", AvgStuScore3)

    // Register connectors
    kis.Pool().CaaSInit("Score12Cache", InitScore12Cache)
}
ログイン後にコピー

Main Process

main.go

package main

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

func RunFlowCalStuAvgScore12(ctx context.Context, flow kis.Flow) error {

    // Commit data
    _ = flow.CommitRow(`{"stu_id":101, "score_1":100, "score_2":90}`)
    _ = flow.CommitRow(`{"stu_id":102, "score_1":100, "score_2":80}`)

    // Run the flow
    if err := flow.Run(ctx); err != nil {
        return err
    }

    return nil
}

func RunFlowCalStuAvgScore3(ctx context.Context, flow kis.Flow) error {

    // Commit data
    _ = flow.CommitRow(`{"stu_id":101, "score_3": 80}`)
    _ = flow.CommitRow(`{"stu_id":102, "score_3": 70}`)

    // Run the flow
    if err := flow.Run(ctx); err != nil {
        return err
    }

    return nil
}

func main() {
    ctx := context.Background()

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

    var wg sync.WaitGroup
    wg.Add(2)

    go func() {
        // Run flow1 concurrently
        defer wg.Done()

        flow1 := kis.Pool().GetFlow("CalStuAvgScore12")
        if flow1 == nil {
            panic("flow1 is nil")
        }

        if err := RunFlowCalStuAvgScore12(ctx, flow1); err != nil {
            panic(err)
        }
    }()

    go func() {
        // Run flow2 concurrently
        defer wg.Done()

        flow2 := kis.Pool().GetFlow("CalStuAvgScore3")
        if flow2 == nil {
            panic("flow2 is nil")
        }

        if err := RunFlowCalStuAvgScore3(ctx, flow2); err != nil {
            panic(err)
        }
    }()

    wg.Wait()

    return
}
ログイン後にコピー

Two Goroutines are launched concurrently to execute Flow1 and Flow2, calculating the final average scores for student 101 and student 102.

Execution Results 

===> Call Connector InitScore12Cache
Connected to Redis: PONG
Add FlowRouter FlowName=CalStuAvgScore12
===> Call Connector InitScore12Cache
Connected to Redis: PONG
Add FlowRouter FlowName=CalStuAvgScore3
->Call Func VerifyStu
->Call Func VerifyStu
->Call Func AvgStuScore12
->Call Func LoadScoreAvg12
->Call Func SaveScoreAvg12
->Call Func PrintStuAvgScore, in Flow[CalStuAvgScore12]
stuid: [101], avg score: [95]
stuid: [102], avg score: [90]
->Call Func AvgStuScore3
->Call Func PrintStuAvgScore, in Flow[CalStuAvgScore3]
stuid: [101], avg score: [90]
stuid: [102], avg score: [83.33333333333333]
ログイン後にコピー

In Flow[CalStuAvgScore3], we observe the final computed average scores for scores 1, 2, and 3.

Author: Aceld
GitHub: https://github.com/aceld

KisFlow Open Source Project Address: https://github.com/aceld/kis-flow

Document: https://github.com/aceld/kis-flow/wiki

Part1-OverView
Part2.1-Project Construction / Basic Modules
Part2.2-Project Construction / Basic Modules
Part3-Data Stream
Part4-Function Scheduling
Part5-Connector
Part6-Configuration Import and Export
Part7-KisFlow Action
Part8-Cache/Params Data Caching and Data Parameters
Part9-Multiple Copies of Flow
Part10-Prometheus Metrics Statistics
Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection

Case1-Quick Start
Case2-Flow Parallel Operation
Case3-Application of KisFlow in Multi-Goroutines
Case4-KisFlow in Message Queue (MQ) Applications

以上がケース (II) - KisFlow-Golang ストリーム リアルタイム コンピューティング - フロー並列操作の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

ソース：dev.to
前の記事：ビルダーのデザインパターン 次の記事：ハーロックv.リリース済み
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
最新の問題
function_exists() はカスタム関数を決定できません Function test () {return true;} if (function_exists ('test')) {echo "テストは関数です";
から 2024-04-29 11:01:01
0
2
1369
Google Chromeのモバイル版を表示する方法 こんにちは、先生、Google Chrome をモバイル版に変更するにはどうすればよいですか?
から 2024-04-23 00:22:19
0
9
1463
子ウィンドウは親ウィンドウを操作しますが、出力は応答しません。 最初の 2 つの文は実行可能ですが、最後の文は実装できません。
から 2024-04-19 15:37:47
0
1
1292
親ウィンドウには出力がありません document.onclick = function(){ window.opener.document.write('私は子ウィンドウの出力です');
から 2024-04-18 23:52:34
0
1
1185
CSS マインド マッピングに関するコースウェアはどこにありますか? コースウェア
から 2024-04-16 10:10:18
0
0
1262
関連トピック
詳細>
人気のおすすめ
人気のチュートリアル
詳細>
関連するチュートリアル
人気のおすすめ
最新のコース
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート
私たちについて 免責事項 Sitemap
PHP中国語ウェブサイト：福祉オンライン PHP トレーニング，PHP 学習者の迅速な成長を支援します！