package consume import ( "applet/app/cfg" "applet/app/e" "applet/app/lib/wechat" md2 "applet/app/md" "applet/app/utils" "applet/app/utils/cache" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles" db "code.fnuoos.com/zhimeng/model.git/src" implement2 "code.fnuoos.com/zhimeng/model.git/src/implement" "code.fnuoos.com/zhimeng/model.git/src/super/implement" "code.fnuoos.com/zhimeng/model.git/src/super/model" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" "time" ) func AdOriginalDataApplication(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>AdOriginalDataApplication>>>>>>>>>>>>") 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) one_circles.Init(cfg.RedisAddr) var res amqp.Delivery var ok bool for { res, ok = <-delivery if ok == true { err = handleAdOriginalDataApplication(res.Body) fmt.Println("err ::: ", err) err = res.Ack(true) } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleAdOriginalDataApplication(msgData []byte) error { //1、解析mq中queue的数据结构体 var msg md2.ZhiosAdOriginalDataApplication var tmpString string err := json.Unmarshal(msgData, &tmpString) if err != nil { fmt.Println(err.Error()) return err } fmt.Println(tmpString) err = json.Unmarshal([]byte(tmpString), &msg) if err != nil { return err } if msg.IsEnd == "1" { cache.SetEx(msg.Mid+":original.wx.ad.data", "0", 5) } time.Sleep(time.Microsecond * 100) // 等待100毫秒 fmt.Println("handleAdOriginalDataApplication:::::::::::>>>>>>>>>") fmt.Println(msg) if db.DBs[msg.Mid] == nil { return nil } eg := db.DBs[msg.Mid] wxApiService, err := wechat.NewWxApiService(msg.Mid, msg.ComponentAppid, msg.ComponentAppsecret) if err != nil { return err } userWxAppletListDb := implement.NewUserWxAppletListDb(db.Db) UserWxAppletList, err := userWxAppletListDb.GetUserWxAppletList(msg.Mid) if err != nil { return err } if UserWxAppletList == nil { return e.NewErr(400, "未查询到对应记录") } err, args := wxApiService.GetAdposDetail(UserWxAppletList.Appid, 1, 1, msg.Date, msg.Date, msg.AdId) if err != nil { return err } NewAppletApplicationAdSpaceListDb := implement2.NewAppletApplicationAdSpaceListDb(eg) for _, v := range args.List { count, _ := db.Db.Where("uuid=? and date=? and app_id=? and slot_id=?", msg.Mid, msg.Date, v.AppId, v.AdUnitId).Count(&model.OriginalWxAdData{}) if count > 0 { continue } adData, _ := NewAppletApplicationAdSpaceListDb.GetAppletApplicationAdSpaceListByAdId(v.AdUnitId) if adData == nil { continue } tmp := model.OriginalWxAdData{ Uuid: utils.StrToInt(msg.Mid), MediumId: adData.MediumId, AppId: adData.AppId, PlatformAppId: v.AppId, SlotId: v.AdUnitId, AdSlot: v.StatItem.AdSlot, Date: msg.Date, ReqSuccCount: int(v.StatItem.ReqSuccCount), ExposureCount: int(v.StatItem.ExposureCount), ExposureRate: utils.Float64ToStr(v.StatItem.ExposureRate), ClickCount: int(v.StatItem.ClickCount), ClickRate: utils.Float64ToStr(v.StatItem.ClickRate), PublisherIncome: int(v.StatItem.PublisherIncome), Ecpm: utils.Float64ToStr(v.StatItem.Ecpm), CreateAt: time.Now().Format("2006-01-02 15:04:05"), UpdateAt: time.Now().Format("2006-01-02 15:04:05"), Platform: "wx_applet", } db.Db.Insert(&tmp) } return nil }