蛋蛋星球RabbitMq消费项目
25'ten fazla konu seçemezsiniz Konular bir harf veya rakamla başlamalı, kısa çizgiler ('-') içerebilir ve en fazla 35 karakter uzunluğunda olabilir.
 
 
 
 
 
 

160 satır
4.3 KiB

  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. db2 "code.fnuoos.com/EggPlanet/egg_models.git/src"
  9. "code.fnuoos.com/EggPlanet/egg_models.git/src/model"
  10. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  11. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  12. "encoding/json"
  13. "errors"
  14. "fmt"
  15. "github.com/jinzhu/copier"
  16. "github.com/streadway/amqp"
  17. "time"
  18. "xorm.io/xorm"
  19. )
  20. func UserDeleteConsume(queue md.MqQueue) {
  21. var backCfg db2.BackUpDBCfg
  22. copier.Copy(&backCfg, &cfg.BackUpDb)
  23. backUpDb, err := db2.InitBackUpDB(&backCfg)
  24. if err != nil {
  25. logx.Error(err)
  26. return
  27. }
  28. fmt.Println(">>>>>>>>>>>>UserDeleteConsume>>>>>>>>>>>>")
  29. ch, err := rabbit.Cfg.Pool.GetChannel()
  30. if err != nil {
  31. logx.Error(err)
  32. return
  33. }
  34. defer ch.Release()
  35. //1、将自己绑定到交换机上
  36. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  37. //2、取出数据进行消费
  38. ch.Qos(100)
  39. delivery := ch.Consume(queue.Name, false)
  40. egg_system_rules.Init(cfg.RedisAddr)
  41. var res amqp.Delivery
  42. var ok bool
  43. for {
  44. res, ok = <-delivery
  45. if ok == true {
  46. err = handleUserDeleteConsume(backUpDb, ch, res.Body)
  47. if err != nil {
  48. fmt.Println("UserDeleteConsume_ERR:::::", err.Error())
  49. utils2.FilePutContents("UserDeleteConsume_ERR", utils2.SerializeStr(map[string]interface{}{
  50. "body": res.Body,
  51. "err": err.Error(),
  52. }))
  53. _ = res.Reject(false)
  54. //TODO::重新推回队列末尾,避免造成队列堵塞
  55. var msg *md.CommUserId
  56. json.Unmarshal(res.Body, &msg)
  57. ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
  58. } else {
  59. //_ = res.Reject(false)
  60. err = res.Ack(true)
  61. }
  62. fmt.Println("err ::: ", err)
  63. } else {
  64. panic(errors.New("error getting message"))
  65. }
  66. }
  67. fmt.Println("get msg done")
  68. }
  69. func handleUserDeleteConsume(backEg *xorm.Engine, ch *rabbit.Channel, msgData []byte) error {
  70. if backEg == nil {
  71. return nil
  72. }
  73. time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
  74. // 1.解析mq中queue的数据结构体
  75. var msg *md.CommUserId
  76. err := json.Unmarshal(msgData, &msg)
  77. if err != nil {
  78. return err
  79. }
  80. eg := db.Db
  81. //1.用户信息
  82. var user model.User
  83. exist, err := eg.Where("id=?", msg.Uid).Get(&user)
  84. if exist {
  85. backEg.Insert(&user)
  86. user.Phone = ""
  87. user.State = 2
  88. user.Nickname = "注销用户"
  89. db.Db.Where("id=?", user.Id).Cols("state,phone,nickname").Update(&user)
  90. }
  91. //2.用户关系链
  92. //查出所有下级
  93. var relate []model.UserRelate
  94. err = eg.Where("parent_uid=?", msg.Uid).Find(&relate)
  95. if len(relate) > 0 {
  96. backEg.Insert(&relate)
  97. }
  98. var relateParent []model.UserRelate
  99. err = eg.Where("uid=?", msg.Uid).Find(&relateParent)
  100. var relateParent1 []model.UserRelate
  101. err = eg.Where("uid=?", msg.ParentUid).Find(&relateParent1)
  102. var parentIds = make([]int64, 0)
  103. if len(relateParent) > 0 {
  104. backEg.Insert(&relateParent)
  105. for _, v := range relateParent {
  106. parentIds = append(parentIds, v.ParentUid)
  107. }
  108. }
  109. var ids = make([]int64, 0)
  110. for _, v := range relate {
  111. ids = append(ids, v.Uid)
  112. }
  113. //读出所有下级重新处理
  114. var allRelate []model.UserRelate
  115. err = eg.In("uid", ids).Desc("level").Find(&allRelate)
  116. for _, v := range allRelate {
  117. if utils2.InArrInt64(v.ParentUid, parentIds) {
  118. eg.Where("id=?", v.Id).Delete(&model.UserRelate{})
  119. continue
  120. }
  121. if v.ParentUid == utils2.StrToInt64(msg.Uid) && utils2.StrToInt64(msg.ParentUid) > 0 { //如果相等
  122. level := v.Level
  123. var tmp = model.UserRelate{
  124. ParentUid: utils2.StrToInt64(msg.ParentUid),
  125. Uid: v.Uid,
  126. Level: level,
  127. InviteTime: v.InviteTime,
  128. }
  129. eg.Insert(&tmp)
  130. for _, v1 := range relateParent1 {
  131. level++
  132. var tmp1 = model.UserRelate{
  133. ParentUid: v1.ParentUid,
  134. Uid: v.Uid,
  135. Level: level,
  136. InviteTime: v.InviteTime,
  137. }
  138. eg.Insert(&tmp1)
  139. }
  140. }
  141. }
  142. eg.Where("parent_uid=?", msg.Uid).Cols("parent_uid").Update(&model.User{ParentUid: utils2.StrToInt64(msg.ParentUid)})
  143. _, err = eg.Where("parent_uid=?", msg.Uid).Delete(&model.UserRelate{})
  144. _, err = eg.Where("uid=?", msg.Uid).Delete(&model.UserRelate{})
  145. var UserRealNameAuth model.UserRealNameAuth
  146. exist, err = eg.Where("uid=?", msg.Uid).Get(&UserRealNameAuth)
  147. if exist {
  148. backEg.Insert(&UserRealNameAuth)
  149. }
  150. eg.Where("uid=?", msg.Uid).Delete(&model.UserRealNameAuth{})
  151. return nil
  152. }