环节处理改为批处理

et_cache/cacheHandler.go

@@ -5,10 +5,9 @@ import (
 	"fmt"
 	"gitea.anxinyun.cn/container/common_models"
 	"gitea.anxinyun.cn/container/common_models/constant/redisKey"
-	"gitea.anxinyun.cn/container/common_utils"
-	"gitea.anxinyun.cn/container/common_utils/configLoad"
 	"log"
 	"node/stages"
+	"sync"
 )
 
 type CacheHandler struct {
@@ -16,19 +15,33 @@ type CacheHandler struct {
 	stage       *stages.Stage
 }
 
-func NewCacheHandler() *CacheHandler {
-	redisAddr := configLoad.LoadConfig().GetString("redis.address")
-	configHelper := common_utils.NewConfigHelper(redisAddr)
+func NewCacheHandler(cacheServer *cacheSer.CacheServer) *CacheHandler {
+	//redisAddr := configLoad.LoadConfig().GetString("redis.address")
+	//configHelper := common_utils.NewConfigHelper(redisAddr)
 	the := &CacheHandler{
-		stage:       stages.NewStage("测点数据缓存"),
-		cacheServer: cacheSer.NewCacheServer(configHelper),
+		stage:       stages.NewStage("滑窗过滤"),
+		cacheServer: cacheServer, //cacheSer.NewCacheServer(configHelper),
 	}
-	the.stage.AddProcess(the.enqueue)
+	the.stage.AddProcess(the.enqueueForStations)
 	return the
 }
 func (the *CacheHandler) GetStage() stages.Stage {
 	return *the.stage
 }
+
+func (the *CacheHandler) enqueueForStations(data []*common_models.ProcessData) []*common_models.ProcessData {
+	var wg sync.WaitGroup // 初始化 WaitGroup
+	for _, processData := range data {
+		wg.Add(1)
+		go func(pd *common_models.ProcessData) {
+			defer wg.Done()
+			the.enqueue(pd)
+		}(processData)
+	}
+	wg.Wait()
+	return data
+}
+
 func (the *CacheHandler) enqueue(p *common_models.ProcessData) *common_models.ProcessData {
 
 	for _, station := range p.Stations {