| | |
| | | |
| | | var msg []byte |
| | | var err error |
| | | |
| | | msgChan := make(chan []byte,100) |
| | | for { |
| | | select { |
| | | // case <-ctx.Done(): |
| | | // return |
| | | default: |
| | | msg, err = c.Recv() |
| | | msgChan <- msg |
| | | if err != nil { |
| | | logger.Info("recv error : ", err) |
| | | fmt.Println("recv error : ", err) |
| | | continue |
| | | } else { |
| | | wg := sync.WaitGroup{} |
| | | for msg := range msgChan { |
| | | wg.Add(1) |
| | | go func() { |
| | | logger.Debug("当前时间戳:",time.Now().Unix()) |
| | | arg := ruleserver.SdkDatas{} |
| | | //paramFormat(msg, &arg) |
| | | logger.Error() |
| | | start := time.Now() |
| | | m := paramFormat(msg, &arg) |
| | | // 进行规则处理判断(打上规则的标签) |
| | | ruleserver.Judge(&arg,&m) // 把sdkMessage传进去,方便缓存数据时拼出一个resultMag |
| | | // 把arg里的打的标签拿出来给m再封装一层 |
| | | resultMsg := ruleserver.ResultMsg{SdkMessage: &m, RuleResult: arg.RuleResult} |
| | | logger.Debug("比对完所用时间:",time.Since(start)) |
| | | logger.Debug("规则判断完所用时间:",time.Since(start)) |
| | | // 将打完标签的数据插入到ES |
| | | insertdata.InsertToEs(resultMsg) |
| | | logger.Debug("插入完Es所用时间:",time.Since(start)) |
| | | //事件推送 |
| | | //go labelFilter.Judge(resultMsg) |
| | | wg.Done() |
| | | }() |
| | | } |
| | | } |
| | | } |
| | | } |