コミュニティ 学ぶ ツールライブラリ レジャー
検索
日本語
ホームページ > バックエンド開発 > Golang > 本文

ケース (IV) - KisFlow-Golang Stream Real-メッセージ キュー (MQ) アプリケーションの KisFlow

WBOY
リリース： 2024-07-18 03:44:47
オリジナル
874 人が閲覧しました

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-プロジェクト構築/基本モジュール
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 開発者ドキュメント

KisFlow と Kafka

サンプルソースコード

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

この例では、サードパーティの Kafka クライアント SDK として github.com/segmentio/kafka-go を使用します (開発者は他の 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)
}
ログイン後にコピー

Nsq を使用した KisFlow

サンプルソースコード:

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

この KisFlow コンシューマーは、サードパーティ SDK として github.com/nsqio/go-nsq を使用します。

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)
}
ログイン後にコピー

RocketMQ を使用した KisFlow

サンプルソースコード:

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

RocketMQ コンシューマー SDK として github.com/apache/rocketmq-client-go を使用します。

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

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

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

ソース：dev.to
前の記事：日常のタスクのベンチマークでベストを測定する際の Go と Rust の比較 次の記事：Go のガベージ コレクターについて理解する: 詳細ガイド
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
最新の問題
function_exists() はカスタム関数を決定できません Function test () {return true;} if (function_exists ('test')) {echo "テストは関数です";
から 2024-04-29 11:01:01
0
2
1956
Google Chromeのモバイル版を表示する方法 こんにちは、先生、Google Chrome をモバイル版に変更するにはどうすればよいですか?
から 2024-04-23 00:22:19
0
10
2118
子ウィンドウは親ウィンドウを操作しますが、出力は応答しません。 最初の 2 つの文は実行可能ですが、最後の文は実装できません。
から 2024-04-19 15:37:47
0
1
1786
親ウィンドウには出力がありません document.onclick = function(){ window.opener.document.write('私は子ウィンドウの出力です');
から 2024-04-18 23:52:34
0
1
1687
CSS マインド マッピングに関するコースウェアはどこにありますか? コースウェア
から 2024-04-16 10:10:18
0
0
1696
関連トピック
詳細>
人気のおすすめ
人気のチュートリアル
詳細>
関連するチュートリアル
人気のおすすめ
最新のコース
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート