package consume import ( "applet/app/cfg" "applet/app/db" utils2 "applet/app/utils" "applet/app/utils/cache" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/EggPlanet/egg_models.git/src/implement" "code.fnuoos.com/EggPlanet/egg_models.git/src/model" "code.fnuoos.com/EggPlanet/egg_system_rules.git" md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" "time" ) func EggEnergyTeamAssistanceConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>EggEnergyTeamAssistanceConsume>>>>>>>>>>>>") ch, err := rabbit.Cfg.Pool.GetChannel() if err != nil { logx.Error(err) return } defer ch.Release() //1、将自己绑定到交换机上 ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) //2、取出数据进行消费 ch.Qos(2) delivery := ch.Consume(queue.Name, false) egg_system_rules.Init(cfg.RedisAddr) var res amqp.Delivery var ok bool for { res, ok = <-delivery if ok == true { err = handleEggEnergyTeamAssistanceDataConsume(res.Body) if err != nil { fmt.Println("EggEnergyTeamAssistanceConsume_ERR:::::", err.Error()) utils2.FilePutContents("EggEnergyTeamAssistance_ERR", utils2.SerializeStr(map[string]interface{}{ "body": res.Body, "err": err.Error(), })) } //_ = res.Reject(false) err = res.Ack(true) fmt.Println("err ::: ", err) } else { panic(errors.New("error getting message")) } } } func handleEggEnergyTeamAssistanceDataConsume(msgData []byte) error { time.Sleep(time.Duration(10) * time.Millisecond) //休眠100毫秒 // 1.解析mq中queue的数据结构体 var msg *md2.EggEnergyTeamAssistanceReq err := json.Unmarshal(msgData, &msg) if err != nil { return err } // 2.查询用户是否已经助力过 //typesPrefix := "incentive_eggSmash" //callbackDb := implement.NewAdvertisingCallbackDb(db.Db) //count, err := callbackDb.AdvertisingCallbackCount(utils2.Int64ToStr(msg.AssistanceUid), msg.SignStart, msg.SignEnd, typesPrefix) //if err != nil { // return err //} cacheKey := fmt.Sprintf("EggEnergyTeamAssistance:%d", msg.Uid) if !cache.Exists(cacheKey) { targetTime, err1 := time.Parse("2006-01-02 15:04:05", msg.SignEnd) if err1 != nil { fmt.Println("解析时间错误:", err1) return err1 } diff := targetTime.Sub(time.Now()).Seconds() _, err = cache.SetEx(cacheKey, nil, int(diff)) if err != nil { return err } } count, err1 := cache.GetBit(cacheKey, msg.AssistanceUid) if err1 != nil { return err1 } if count == 0 { _, err = cache.SetBit(cacheKey, msg.AssistanceUid, 1) if err != nil { return err } } // 3. 变更数据 assistanceDb := implement.NewEggEnergyTeamAssistanceDb(db.Db) now := time.Now() date := now.Format("2006-01-02") session := db.Db.Where("") teamAssistance, err := assistanceDb.EggEnergyTeamAssistanceGetOneByParamsBySession(session, map[string]interface{}{ "key": "sign_id", "value": msg.SignId, }) if err != nil { return err } if teamAssistance == nil { m := model.EggEnergyTeamAssistance{ Uid: msg.Uid, AssistedNum: 1, AssistedTimes: 1, RewardScore: utils2.Float64ToStr(msg.AssistanceValue), Date: date, SignId: msg.SignId, CreateAt: now.Format("2006-01-02 15:04:05"), UpdateAt: now.Format("2006-01-02 15:04:05"), } _, err = assistanceDb.EggEnergyTeamAssistanceInsert(&m) if err != nil { return err } } else { teamAssistance.RewardScore = utils2.Float64ToStr(msg.AssistanceValue + utils2.StrToFloat64(teamAssistance.RewardScore)) // 如果是第一条说明还没有被记录过, 助力人数 + 1 if count == 0 { teamAssistance.AssistedNum++ } teamAssistance.AssistedTimes++ _, err = assistanceDb.EggEnergyTeamAssistanceUpdate(teamAssistance.Id, teamAssistance, "assisted_num", "reward_score", "assisted_times") if err != nil { return err } } return nil }