package consume import ( "applet/app/cfg" "applet/app/db" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/EggPlanet/egg_models.git/src/model" "code.fnuoos.com/EggPlanet/egg_system_rules.git" "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy" 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 AdvertisingSignConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>JpushRecordConsume>>>>>>>>>>>>") 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(1) 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 = handleAdvertisingSignConsume(res.Body) err = res.Ack(true) fmt.Println("err ::: ", err) } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleAdvertisingSignConsume(msgData []byte) error { time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒 // 1.解析mq中queue的数据结构体 var msg *md.AdvertisingWatch err := json.Unmarshal(msgData, &msg) if err != nil { return err } var data model.AdvertisingCallback db.Db.Where("id=? ", msg.Id).Get(&data) if data.Id == 0 { return errors.New("记录不存在") } if data.IsRun == 1 { return nil } req := md2.HomePageStartSignInReq{ UID: int64(data.Uid), } err = egg_energy.HomePageStartSignIn(db.Db, req) if err != nil { return err } UpdateUserTime(int64(data.Uid), "sign") data.IsRun = 1 db.Db.Where("id=?", data.Id).Cols("is_run").Update(&data) return nil } func UpdateUserTime(uid int64, types string) { count, _ := db.Db.Where("uid=?", uid).Count(&model.UserNoticeTime{}) if count == 0 { tmp := &model.UserNoticeTime{Uid: int(uid)} if types == "login" { tmp.LoginTime = int(time.Now().Unix()) } else { tmp.SignTime = int(time.Now().Unix()) } db.Db.Insert(tmp) } else { tmp := &model.UserNoticeTime{} str := "" if types == "login" { str = "login_time" tmp.LoginTime = int(time.Now().Unix()) } else { str = "sign_time" tmp.SignTime = int(time.Now().Unix()) } db.Db.Where("uid=?", uid).Cols(str).Update(tmp) } }