package consume import ( "applet/app/cfg" "applet/app/db" utils2 "applet/app/utils" "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" zhios_order_relate_utils "code.fnuoos.com/EggPlanet/egg_models.git/utils" "code.fnuoos.com/EggPlanet/egg_system_rules.git" "code.fnuoos.com/EggPlanet/egg_system_rules.git/enum" md3 "code.fnuoos.com/EggPlanet/egg_system_rules.git/md" "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule" md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md" "code.fnuoos.com/EggPlanet/egg_system_rules.git/svc" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" "strconv" "time" ) // EggPublishDataConsume 处理冻结用户流水 func EggPublishDataConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>EggPublishDataConsume>>>>>>>>>>>>") 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 = handleEggPublishDataConsume(res.Body) if err != nil { fmt.Println("EggPublishDataConsume_ERR:::::", err.Error()) utils2.FilePutContents("EggPublishDataConsume_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 handleEggPublishDataConsume(msgData []byte) error { time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒 // 1.解析mq中queue的数据结构体 var msg *md2.EggRoutKeyForPunishmentRecordData err := json.Unmarshal(msgData, &msg) if err != nil { return err } settingDb := implement.NewEggEnergyBasicSettingDb(db.Db) setting, err := settingDb.EggEnergyBasicSettingGetOne() if err != nil { return err } //1、分布式锁阻拦 requestIdPrefix1 := fmt.Sprintf(md2.DealUserCoinRequestIdPrefix, setting.PersonEggEnergyCoinId, msg.Uid) cb1, err := svc.HandleDistributedLock(zhios_order_relate_utils.Int64ToStr(msg.Uid), strconv.Itoa(setting.PersonEggEnergyCoinId), requestIdPrefix1) if err != nil { return err } if cb1 != nil { defer cb1() // 释放锁 } requestIdPrefix2 := fmt.Sprintf(md2.DealUserCoinRequestIdPrefix, setting.TeamEggEnergyCoinId, msg.Uid) cb2, err := svc.HandleDistributedLock(zhios_order_relate_utils.Int64ToStr(msg.Uid), strconv.Itoa(setting.TeamEggEnergyCoinId), requestIdPrefix2) if err != nil { return err } if cb2 != nil { defer cb2() // 释放锁 } session := db.Db.NewSession() defer session.Close() session.Begin() var userVirtualWallet model.UserVirtualAmount get, err := session.Table("user_virtual_amount"). Where("uid = ?", msg.Uid). And("coin_id = ?", setting.TeamEggEnergyCoinId).Get(&userVirtualWallet) if err != nil { fmt.Println(err) return err } if !get { fmt.Println("userVirtualWallet1 not exist ======>", msg.Uid) return err } if utils2.StrToFloat64(msg.Amount) <= utils2.StrToFloat64(userVirtualWallet.Amount) { // 待结够扣 err = rule.DealUserVirtualCoinMinus(session, md3.DealUserVirtualCoinReq{ Kind: "sub", Title: enum.EggEnergyMaliciousAccountFlashingDeductionTeam.String(), TransferType: int(enum.EggEnergyMaliciousAccountFlashingDeductionTeam), CoinId: setting.TeamEggEnergyCoinId, Uid: msg.Uid, Amount: utils2.StrToFloat64(msg.Amount), }) if err != nil { return err } } else { // 待结不够扣 // 扣完当前待结余额 if utils2.StrToFloat64(userVirtualWallet.Amount) > 0 { err = rule.DealUserVirtualCoinMinus(session, md3.DealUserVirtualCoinReq{ Kind: "sub", Title: enum.EggEnergyMaliciousAccountFlashingDeductionTeam.String(), TransferType: int(enum.EggEnergyMaliciousAccountFlashingDeductionTeam), CoinId: setting.TeamEggEnergyCoinId, Uid: msg.Uid, Amount: utils2.StrToFloat64(userVirtualWallet.Amount), }) } remainingAmount := utils2.StrToFloat64(msg.Amount) - utils2.StrToFloat64(userVirtualWallet.Amount) err := rule.DealUserVirtualCoinMinus(session, md3.DealUserVirtualCoinReq{ Kind: "sub", Title: enum.EggEnergyMaliciousAccountFlashingDeductionPersonal.String(), TransferType: int(enum.EggEnergyMaliciousAccountFlashingDeductionPersonal), CoinId: setting.PersonEggEnergyCoinId, Uid: msg.Uid, Amount: remainingAmount, }) if err != nil { return err } } _, err = session.Table("punishment_record"). Where("uid = ?", msg.Uid). Update(model.PunishmentRecord{ Complete: 1, }) if err != nil { return err } session.Commit() fmt.Println("handleEggPublishDataConsume_succeed ====>", msg.Uid) return nil }