蛋蛋星球RabbitMq消费项目
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

egg_energy_team_assistance_consume.go 3.3 KiB

3 days ago
3 days ago
3 days ago
1 day ago
3 days ago
3 days ago
1 day ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
3 days ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. package consume
  2. import (
  3. "applet/app/cfg"
  4. "applet/app/db"
  5. utils2 "applet/app/utils"
  6. "applet/app/utils/logx"
  7. "applet/consume/md"
  8. "code.fnuoos.com/EggPlanet/egg_models.git/src/implement"
  9. "code.fnuoos.com/EggPlanet/egg_models.git/src/model"
  10. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  11. md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md"
  12. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  13. "encoding/json"
  14. "errors"
  15. "fmt"
  16. "github.com/streadway/amqp"
  17. "time"
  18. )
  19. func EggEnergyTeamAssistanceConsume(queue md.MqQueue) {
  20. fmt.Println(">>>>>>>>>>>>EggEnergyTeamAssistanceConsume>>>>>>>>>>>>")
  21. ch, err := rabbit.Cfg.Pool.GetChannel()
  22. if err != nil {
  23. logx.Error(err)
  24. return
  25. }
  26. defer ch.Release()
  27. //1、将自己绑定到交换机上
  28. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  29. //2、取出数据进行消费
  30. ch.Qos(1)
  31. delivery := ch.Consume(queue.Name, false)
  32. egg_system_rules.Init(cfg.RedisAddr)
  33. var res amqp.Delivery
  34. var ok bool
  35. for {
  36. res, ok = <-delivery
  37. if ok == true {
  38. err = handleEggEnergyTeamAssistanceDataConsume(res.Body)
  39. if err != nil {
  40. fmt.Println("EggEnergyTeamAssistanceConsume_ERR:::::", err.Error())
  41. utils2.FilePutContents("EggEnergyTeamAssistance_ERR", utils2.SerializeStr(map[string]interface{}{
  42. "body": res.Body,
  43. "err": err.Error(),
  44. }))
  45. }
  46. //_ = res.Reject(false)
  47. err = res.Ack(true)
  48. fmt.Println("err ::: ", err)
  49. } else {
  50. panic(errors.New("error getting message"))
  51. }
  52. }
  53. }
  54. func handleEggEnergyTeamAssistanceDataConsume(msgData []byte) error {
  55. time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
  56. // 1.解析mq中queue的数据结构体
  57. var msg *md2.EggEnergyTeamAssistanceReq
  58. err := json.Unmarshal(msgData, &msg)
  59. if err != nil {
  60. return err
  61. }
  62. // 2.查询用户是否已经助力过
  63. typesPrefix := "incentive_eggSmash"
  64. callbackDb := implement.NewAdvertisingCallbackDb(db.Db)
  65. count, err := callbackDb.AdvertisingCallbackCount(utils2.Int64ToStr(msg.AssistanceUid), msg.SignStart, msg.SignEnd, typesPrefix)
  66. if err != nil {
  67. return err
  68. }
  69. // 3. 变更数据
  70. assistanceDb := implement.NewEggEnergyTeamAssistanceDb(db.Db)
  71. now := time.Now()
  72. date := now.Format("2006-01-02")
  73. session := db.Db.Where("")
  74. teamAssistance, err := assistanceDb.EggEnergyTeamAssistanceGetOneByParamsBySession(session, map[string]interface{}{
  75. "key": "sign_id",
  76. "value": msg.SignId,
  77. })
  78. if err != nil {
  79. return err
  80. }
  81. if teamAssistance == nil {
  82. m := model.EggEnergyTeamAssistance{
  83. Uid: msg.Uid,
  84. AssistedNum: 1,
  85. AssistedTimes: 1,
  86. RewardScore: utils2.Float64ToStr(msg.AssistanceValue),
  87. Date: date,
  88. SignId: msg.SignId,
  89. CreateAt: now.Format("2006-01-02 15:04:05"),
  90. UpdateAt: now.Format("2006-01-02 15:04:05"),
  91. }
  92. _, err = assistanceDb.EggEnergyTeamAssistanceInsert(&m)
  93. if err != nil {
  94. return err
  95. }
  96. } else {
  97. teamAssistance.RewardScore = utils2.Float64ToStr(msg.AssistanceValue + utils2.StrToFloat64(teamAssistance.RewardScore))
  98. // 如果是第一条说明还没有被记录过, 助力人数 + 1
  99. if count == 1 {
  100. teamAssistance.AssistedNum++
  101. }
  102. teamAssistance.AssistedTimes++
  103. _, err = assistanceDb.EggEnergyTeamAssistanceUpdate(teamAssistance.Id, teamAssistance, "assisted_num", "reward_score", "assisted_times")
  104. if err != nil {
  105. return err
  106. }
  107. }
  108. return nil
  109. }