Kes (II) - Pengkomputeran Masa Nyata Aliran KisFlow-Golang

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

Github: https://github.com/aceld/kis-flow
Dokumen: https://github.com/acceld/kis-flow/wiki

Bahagian1-Paparan Keseluruhan
Bahagian2.1-Pembinaan Projek / Modul Asas
Bahagian2.2-Pembinaan Projek / Modul Asas
Bahagian3-Strim Data
Bahagian4-Penjadualan Fungsi
Bahagian5-Penyambung
Bahagian6-Konfigurasi Import dan Eksport
Bahagian7-KisFlow Action
Bahagian8-Cache/Params Caching Data dan Parameter Data
Bahagian9-Berbilang Salinan Aliran
Bahagian10-Statistik Metrik Prometheus
Bahagian11-Pendaftaran Adaptif Jenis Parameter FaaS Berdasarkan Refleksi

Kes1-Permulaan Pantas
Kes2-Kendalian Selari Aliran
Kes3-Penggunaan KisFlow dalam Pelbagai Goroutine
Case4-KisFlow dalam Aplikasi Baris Mesej (MQ)

Muat turun Sumber KisFlow

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

Salin selepas log masuk

Dokumentasi Pembangun KisFlow

Contoh Kod Sumber

https://github.com/acceld/kis-flow-usage/tree/main/8-connector

KisFlow Boleh Mencapai Gabungan Dua Aliran melalui Penyambung

Menggunakan gabungan dua aliran berikut, pengenalan ini akan merangkumi antara muka dan penggunaan Penyambung.

Rajah Aliran Data

Flow Diagram

Pengenalan Kes

Andaikan pelajar mempunyai empat sifat:

Student ID: stu_id
Credit 1: score_1
Credit 2: score_2
Credit 3: score_3

Salin selepas log masuk

Tentukan Aliran1: CalStuAvgScore-1-2 untuk mengira purata skor pelajar bagi Kredit 1 (skor_1) dan Kredit 2 (skor_2) (purata_skor_1_2).
Tentukan Aliran2: CalStuAvgScore-3 untuk mengira purata skor pelajar bagi Kredit 3 (skor_3) dan purata_skor_1_2, iaitu purata Kredit 1, Kredit 2 dan Kredit 3. Purata Kredit 1 dan Kredit 2 disediakan oleh Flow1.

Aliran1

Flow1 terdiri daripada 4 fungsi:

V (Fungsi: VerifyStu) untuk mengesahkan kesahihan StuId
C (Fungsi: AvgStuScore12) untuk mengira skor purata Kredit 1 dan Kredit 2
S (Fungsi: SaveScoreAvg12) untuk menyimpan avg_score_1_2 dalam Redis
E (Fungsi: PrintStuAvgScore) untuk mencetak purata skor Kredit 1 dan Kredit 2.

Aliran2

Flow2 terdiri daripada 4 fungsi:

V (Fungsi: VerifyStu) untuk mengesahkan kesahihan StuId
L (Fungsi: LoadScoreAvg12) untuk membaca purata skor pelajar semasa bagi Kredit 1 dan Kredit 2 (avg_skor_1_2) yang dikira oleh Flow1
C (Fungsi: AvgStuScore3) untuk mengira purata skor Kredit 3 dan purata skor Kredit 1 dan Kredit 2
E (Fungsi: PrintStuAvgScore) untuk mencetak purata skor Kredit 1, Kredit 2 dan Kredit 3.

conf/func/func-AvgStuScore-3.yml

kistype: func
fname: AvgStuScore3
fmode: Calculate
source:
    name: SourceStuScore
    must:
        - stu_id

Salin selepas log masuk

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

kistype: func
fname: LoadScoreAvg12
fmode: Load
source:
    name: SourceStuScore
    must:
        - stu_id
option:
    cname: Score12Cache

Salin selepas log masuk

Protokol Data Asas

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"`
}

Salin selepas log masuk

Init Penyambung

Penyambung yang ditakrifkan dalam projek ini, Score12Cache, ialah sumber pautan yang dikaitkan dengan Redis. Penyambung ini memerlukan kaedah permulaan untuk mewujudkan sambungan apabila KisFlow bermula.

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
}

Salin selepas log masuk

Di sini, tika Redis yang berjaya disambungkan disimpan dalam pembolehubah cache penyambung "rdb."

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

Salin selepas log masuk

Pelaksanaan FaaS

Fungsi(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)
}

Salin selepas log masuk

VerifyStu() digunakan untuk mengesahkan data. Jika data tidak memenuhi keperluan, aliran data semasa ditamatkan. Akhirnya, data digunakan semula dan dihantar ke lapisan seterusnya melalui aliran. Seterusnya(kis.ActionDataReuse).

Fungsi(C): AvgStuScore12

faas_avg_skor_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()
}

Salin selepas log masuk

AvgStuScore12() mengira purata skor skor_1 dan skor_2, menghasilkan purata_skor.

Fungsi(S): 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)
}

Salin selepas log masuk

SaveScoreAvg12() menyimpan data dalam Redis melalui Penyambung terikat, menggunakan kekunci yang dikonfigurasikan dalam Penyambung. Akhir sekali, data sumber dihantar secara telus ke fungsi seterusnya.

Fungsi(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()
}

Salin selepas log masuk

PrintStuAvgScore() mencetak purata skor pelajar semasa.

Fungsi(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()
}

Salin selepas log masuk

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()
}

Salin selepas log masuk

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)
}

Salin selepas log masuk

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
}

Salin selepas log masuk

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]

Salin selepas log masuk

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

Atas ialah kandungan terperinci Kes (II) - Pengkomputeran Masa Nyata Aliran KisFlow-Golang - Operasi Selari Aliran. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!