Case (IV) - KisFlow-Golang Stream Real- KisFlow in Message Queue (MQ) Applications
Tags: go
Published: 2024-07-11 02:23:33

<img width="150px" src="https://github.com/aceld/kis-flow/assets/7778936/8729d750-897c-4ba3-98b4-c346188d034e" />

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

---

[Part1-OverView](https://dev.to/aceld/part-1-golang-framework-hands-on-kisflow-streaming-computing-framework-overview-8fh)
[Part2.1-Project Construction / Basic Modules](https://dev.to/aceld/part-2-golang-framework-hands-on-kisflow-streaming-computing-framework-project-construction-basic-modules-cia)
[Part2.2-Project Construction / Basic Modules](https://dev.to/aceld/part-3golang-framework-hands-on-kisflow-stream-computing-framework-project-construction-basic-modules-1epb)
[Part3-Data Stream](https://dev.to/aceld/part-4golang-framework-hands-on-kisflow-stream-computing-framework-data-stream-1mbd)
[Part4-Function Scheduling](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-function-scheduling-4p0h)
[Part5-Connector](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-connector-hcd)
[Part6-Configuration Import and Export](https://dev.to/aceld/part-6golang-framework-hands-on-kisflow-stream-computing-framework-configuration-import-and-export-47o1)
[Part7-KisFlow Action](https://dev.to/aceld/part-7golang-framework-hands-on-kisflow-stream-computing-framework-kisflow-action-3n05)
[Part8-Cache/Params Data Caching and Data Parameters](https://dev.to/aceld/part-8golang-framework-hands-on-cacheparams-data-caching-and-data-parameters-5df5)
[Part9-Multiple Copies of Flow](https://dev.to/aceld/part-8golang-framework-hands-on-multiple-copies-of-flow-c4k)
[Part10-Prometheus Metrics Statistics](https://dev.to/aceld/part-10golang-framework-hands-on-prometheus-metrics-statistics-22f0)
[Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection](https://dev.to/aceld/part-11golang-framework-hands-on-adaptive-registration-of-faas-parameter-types-based-on-reflection-15i9)

---

[Case1-Quick Start](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-quick-start-guide-f51)
[Case2-Flow Parallel Operation](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-flow-parallel-operation-364m)
[Case3-Application of KisFlow in Multi-Goroutines](https://dev.to/aceld/case-iii-kisflow-golang-stream-real-application-of-kisflow-in-multi-goroutines-4m7g)
[Case4-KisFlow in Message Queue (MQ) Applications](https://dev.to/aceld/case-iv-kisflow-golang-stream-real--4k3e)



## Download KisFlow Source

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

[KisFlow Developer Documentation](https://github.com/aceld/kis-flow/wiki)


## KisFlow with Kafka


### Sample source code
https://github.com/aceld/kis-flow-usage/tree/main/12-with_kafka


In this example, we use `github.com/segmentio/kafka-go` as the third-party Kafka Client SDK (developers can choose other Kafka Go tools).

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

## KisFlow with Nsq

### Sample source code:
https://github.com/aceld/kis-flow-usage/tree/main/13-with_nsq

This KisFlow consumer uses `github.com/nsqio/go-nsq` as the third-party SDK.

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


##KisFlow with RocketMQ

###Sample source code:
https://github.com/aceld/kis-flow-usage/tree/main/14-with_rocketmq

Using `github.com/apache/rocketmq-client-go` as the RocketMQ consumer SDK.

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

---

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](https://dev.to/aceld/part-1-golang-framework-hands-on-kisflow-streaming-computing-framework-overview-8fh)
[Part2.1-Project Construction / Basic Modules](https://dev.to/aceld/part-2-golang-framework-hands-on-kisflow-streaming-computing-framework-project-construction-basic-modules-cia)
[Part2.2-Project Construction / Basic Modules](https://dev.to/aceld/part-3golang-framework-hands-on-kisflow-stream-computing-framework-project-construction-basic-modules-1epb)
[Part3-Data Stream](https://dev.to/aceld/part-4golang-framework-hands-on-kisflow-stream-computing-framework-data-stream-1mbd)
[Part4-Function Scheduling](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-function-scheduling-4p0h)
[Part5-Connector](https://dev.to/aceld/part-5golang-framework-hands-on-kisflow-stream-computing-framework-connector-hcd)
[Part6-Configuration Import and Export](https://dev.to/aceld/part-6golang-framework-hands-on-kisflow-stream-computing-framework-configuration-import-and-export-47o1)
[Part7-KisFlow Action](https://dev.to/aceld/part-7golang-framework-hands-on-kisflow-stream-computing-framework-kisflow-action-3n05)
[Part8-Cache/Params Data Caching and Data Parameters](https://dev.to/aceld/part-8golang-framework-hands-on-cacheparams-data-caching-and-data-parameters-5df5)
[Part9-Multiple Copies of Flow](https://dev.to/aceld/part-8golang-framework-hands-on-multiple-copies-of-flow-c4k)
[Part10-Prometheus Metrics Statistics](https://dev.to/aceld/part-10golang-framework-hands-on-prometheus-metrics-statistics-22f0)
[Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection](https://dev.to/aceld/part-11golang-framework-hands-on-adaptive-registration-of-faas-parameter-types-based-on-reflection-15i9)

---
[Case1-Quick Start](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-quick-start-guide-f51)
[Case2-Flow Parallel Operation](https://dev.to/aceld/case-i-kisflow-golang-stream-real-time-computing-flow-parallel-operation-364m)
[Case3-Application of KisFlow in Multi-Goroutines](https://dev.to/aceld/case-iii-kisflow-golang-stream-real-application-of-kisflow-in-multi-goroutines-4m7g)
[Case4-KisFlow in Message Queue (MQ) Applications](https://dev.to/aceld/case-iv-kisflow-golang-stream-real--4k3e)