蛋蛋星球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.

user_delete_consume.go 6.1 KiB

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