Performance bottleneck encountered during testing

[wangbaobaoget]

I tried to start a workflow and send messages to it every 100 milliseconds, but after five minutes, I found that the workflow's message processing speed was far slower than my sending speed, causing a message pile up. And it was found that a large number of keys appeared in Redis, and over time, there will be more and more keys in Redis
The test cases are as follows：
`
package main

import (
"fmt"
"log"
"time"

v1 "github.com/s8sg/goflow/flow/v1"
v11 "github.com/s8sg/goflow/v1"

)

// Helper function for workflow nodes with logging
func doSomething(data []byte, option map[string][]string) ([]byte, error) {
return []byte(fmt.Sprintf("doSomething "%s"", string(data))), nil
}

// Helper function for workflow nodes with logging
func doSomething2(data []byte, option map[string][]string) ([]byte, error) {
return []byte(fmt.Sprintf("doSomething2 "%s"", string(data))), nil
}

// Helper function for workflow nodes with logging
func doSomething3(data []byte, option map[string][]string) ([]byte, error) {
return []byte(fmt.Sprintf("doSomething3 "%s"", string(data))), nil
}

// Helper function for workflow nodes with logging
func doSomething4(data []byte, option map[string][]string) ([]byte, error) {
log.Printf("Executing doSomething4 with data: %s", string(data))
return []byte(fmt.Sprintf("doSomething4 "%s"", string(data))), nil
}

// Define a workflow with at least three nodes
func DefineMultiNodeWorkflow2(flow *v1.Workflow, context *v1.Context) error {
dag := flow.Dag()
dag.Node("node1", doSomething)
dag.Node("node2", doSomething2)
dag.Node("node3", doSomething3)
dag.Node("node4", doSomething4)
dag.Edge("node1", "node2")
dag.Edge("node2", "node3")
dag.Edge("node3", "node4")
return nil
}

func main() {
fs := v11.FlowService{
Port: 8080,
RedisURL: "127.0.0.1:6379",
OpenTraceUrl: "127.0.0.1:5775",
WorkerConcurrency: 5,
RedisPassword: "",
EnableMonitoring: false,
}
// Register a flow with multiple nodes
flowName := "execution-test-flow"
err := fs.Register(flowName, DefineMultiNodeWorkflow2)
if err != nil {
fmt.Printf("Failed to register flow: %s\n", err.Error())
}

// Start the flow (this initializes the runtime)
i := 0
go func() {
	time.Sleep(10 * time.Second)
	for {
		// Create a runtime request
		i = i + 1
		requestBody := fmt.Sprintf("test data %d", i)
		// 生成唯一的RequestId
		requestId := fmt.Sprintf("req-%d-%d", i, time.Now().UnixNano())
		fmt.Printf("------- data: %s, requestId: %s\n", requestBody, requestId)
		req := &v11.Request{
			RequestId: requestId,
			Body:      []byte(requestBody),
		}
		fs.Execute(flowName, req)
		time.Sleep(100 * time.Millisecond)
	}
}()
err = fs.StartWorker()
if err != nil {
	fmt.Printf("Failed to start flow: %s\n", err.Error())
}

}

`
The test results are as follows：

The keys in Redis are as follows:
keys *--->