From 6c20470bf684ff8ccda3f781169ace614aa8793c Mon Sep 17 00:00:00 2001 From: yfh Date: Fri, 27 Sep 2024 17:10:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86=E6=B5=81=E7=A8=8B=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=EF=BC=9A=E5=88=86=E7=BB=84=E8=AE=A1=E7=AE=97=E3=80=81?= =?UTF-8?q?=E6=B5=8B=E7=82=B9=E9=98=88=E5=80=BC=E5=88=86=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- node/app/app.go | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/node/app/app.go b/node/app/app.go index c598faf..169259a 100644 --- a/node/app/app.go +++ b/node/app/app.go @@ -6,6 +6,7 @@ import ( "et_analyze" "et_cache" "et_calc" + "et_calc/group" "et_print" "et_push" "et_sink" @@ -98,28 +99,22 @@ func addWorkStages(nodeStageManage *stages.StageManager) *stages.StageManager { // 单测点计算 calcHandler := et_calc.NewCalcHandler() nodeStageManage.AddStages(calcHandler.GetStage()) - + // 滑窗过滤 cacheHandler := et_cache.NewCacheHandler() nodeStageManage.AddStages(cacheHandler.GetStage()) + // 测点分组计算 + groupCalcHandler := group.NewGroupCalc() + nodeStageManage.AddStages(groupCalcHandler.GetStage()) // Theme 数据存储 sinkThemeHandler := et_sink.NewSinkThemeHandler() nodeStageManage.AddStages(sinkThemeHandler.GetStage()) - // 测点主题数据的阈值分析 + + // 测点阈值分析 stationAnalyzeHandler := et_analyze.NewThresholdHandler() nodeStageManage.AddStages(stationAnalyzeHandler.GetStage()) - // 测点分组计算 - //groupCalcHandler := group.NewGroupCalc() - //nodeStageManage.AddStages(groupCalcHandler.GetStage()) - // EsGroupTheme 数据存储 - //sinkGroupHandler := et_sink.NewSinkGroupHandler() - //nodeStageManage.AddStages(sinkGroupHandler.GetStage()) - - //// (阈值分析) 测点主题数据的阈值分析 + 分组计算后数据的阈值分析 - //groupAnalyzeHandler := et_analyze.NewThresholdHandler() - //nodeStageManage.AddStages(groupAnalyzeHandler.GetStage()) - + // 数据推送 publishHandler := et_push.NewPushHandler() nodeStageManage.AddStages(publishHandler.GetStage()) return nodeStageManage