golang 的 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.
 
 
 

188 lines
4.4 KiB

  1. package consume
  2. import (
  3. "applet/app/db"
  4. "applet/app/db/model"
  5. "applet/app/utils"
  6. "applet/app/utils/logx"
  7. "applet/consume/md"
  8. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  9. "encoding/json"
  10. "errors"
  11. "fmt"
  12. "github.com/streadway/amqp"
  13. "xorm.io/xorm"
  14. )
  15. func ZhiosValidUser(queue md.MqQueue) {
  16. fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>")
  17. ch, err := rabbit.Cfg.Pool.GetChannel()
  18. if err != nil {
  19. logx.Error(err)
  20. return
  21. }
  22. defer ch.Release()
  23. //1、将自己绑定到交换机上
  24. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  25. //2、取出数据进行消费
  26. ch.Qos(300)
  27. delivery := ch.Consume(queue.Name, false)
  28. var res amqp.Delivery
  29. var ok bool
  30. for {
  31. res, ok = <-delivery
  32. if ok == true {
  33. //fmt.Println(string(res.Body))
  34. fmt.Println(">>>>>>>>>>>>>>>>ZhiosAcquisitionCondition<<<<<<<<<<<<<<<<<<<<<<<<<")
  35. err = handleValidUser(res.Body)
  36. //_ = res.Reject(false)
  37. fmt.Println(err)
  38. _ = res.Ack(true)
  39. } else {
  40. panic(errors.New("error getting message"))
  41. }
  42. }
  43. fmt.Println("get msg done")
  44. }
  45. func handleValidUser(msg []byte) error {
  46. //1、解析canal采集至mq中queue的数据结构体
  47. var canalMsg *md.ZhiosAcquisition
  48. fmt.Println(string(msg))
  49. var tmpString string
  50. err := json.Unmarshal(msg, &tmpString)
  51. if err != nil {
  52. fmt.Println(err.Error())
  53. return err
  54. }
  55. fmt.Println(tmpString)
  56. err = json.Unmarshal([]byte(tmpString), &canalMsg)
  57. if err != nil {
  58. return err
  59. }
  60. mid := canalMsg.Mid
  61. eg := db.DBs[mid]
  62. if eg == nil {
  63. return nil
  64. }
  65. if canalMsg.Uid == "" {
  66. return nil
  67. }
  68. RoutineUpdateUserComm(eg, mid, canalMsg.Uid)
  69. return nil
  70. }
  71. func RoutineUpdateUserComm(eg *xorm.Engine, dbName, uid string) {
  72. if utils.StrToInt(uid) == 0 {
  73. return
  74. }
  75. user, _ := db.UserFindByID(eg, uid)
  76. if user == nil {
  77. return
  78. }
  79. userProfile, _ := db.UserProfileFindByID(eg, uid)
  80. if userProfile == nil {
  81. return
  82. }
  83. if userProfile.IsVerify == 1 {
  84. return
  85. }
  86. validConditionMap := map[string]interface{}{}
  87. checkMap := map[string]interface{}{}
  88. todone := 0
  89. //有效会员 校验
  90. vdata := db.SysCfgGetWithDb(eg, dbName, "valid_member_condition")
  91. if err := json.Unmarshal([]byte(vdata), &validConditionMap); err != nil {
  92. return
  93. logx.Warn(err)
  94. }
  95. for k, v := range validConditionMap {
  96. if v != "" && utils.AnyToFloat64(v) > 0 {
  97. todone++
  98. checkMap[k] = v
  99. }
  100. }
  101. for k, v := range checkMap {
  102. switch k {
  103. case "taskVideo":
  104. sum, _ := eg.Where("uid=? and task_type=? and task_id>0", user.Uid, 3).Sum(&model.TaskVideoNum{}, "count")
  105. if sum >= utils.AnyToFloat64(v) {
  106. todone--
  107. }
  108. case "realCheck":
  109. // 检查实名认证
  110. one, _ := db.GetRealNameAuthByUidWithState(eg, user.Uid, 1)
  111. if one != nil && one.Uid == user.Uid {
  112. todone--
  113. }
  114. case "bindPhone":
  115. // 检查绑定手机号
  116. if user.Phone != "" {
  117. todone--
  118. }
  119. case "goodsCommission":
  120. sqlTpl := `SELECT cast(SUM(LEFT(olr.amount,LENGTH(olr.amount)-2)) as decimal(50,4)) AS amount
  121. FROM ord_list_relate olr
  122. LEFT JOIN ord_list ol ON olr.oid = ol.ord_id
  123. LEFT JOIN privilege_card_ord pco ON olr.oid =pco.ord_id
  124. LEFT JOIN duoyou_ord_list dol ON olr.oid =dol.oid
  125. LEFT JOIN recharge_order ro ON olr.oid =ro.oid
  126. LEFT JOIN playlet_sale_order pso ON olr.oid =pso.custom_oid
  127. WHERE olr.uid = ? AND (ol.state<>4 or pco.state=1 or dol.id>0 or ro.status<>'已退款' or pso.status<>'订单退款');
  128. `
  129. todayResult, err := db.QueryNativeString(eg, sqlTpl, uid)
  130. today := "0"
  131. if err == nil {
  132. today = todayResult[0]["amount"]
  133. }
  134. // 累计佣金
  135. if utils.StrToFloat64(today) >= utils.AnyToFloat64(v) {
  136. todone--
  137. }
  138. case "orderPay":
  139. // 付款订单满足v元
  140. ms, err := db.OrderListByUIDByPaidPrice(eg, user.Uid, v, []string{"0", "1", "2", "3", "5"})
  141. if err != nil {
  142. logx.Warn(err)
  143. }
  144. mss := *ms
  145. if len(mss) > 0 {
  146. todone--
  147. }
  148. case "receive":
  149. // 已收货
  150. ms, err := db.OrderListByUIDByState(eg, user.Uid, []string{"1", "2", "3", "5"})
  151. if err != nil {
  152. logx.Warn(err)
  153. }
  154. mss := *ms
  155. if len(mss) > 0 {
  156. todone--
  157. }
  158. case "tbAuth":
  159. // 是否淘宝授权
  160. if userProfile.AccTaobaoAuthTime != 0 {
  161. todone--
  162. }
  163. case "withdraw":
  164. // 提现
  165. sum, err := db.UserWithDrawApplySumByState(eg, user.Uid, "1", "2")
  166. if err != nil {
  167. logx.Warn(err)
  168. }
  169. if sum > 0 {
  170. todone--
  171. }
  172. }
  173. }
  174. // 满足条件则将改用户打为有效用户
  175. if todone == 0 {
  176. userProfile.IsVerify = 1
  177. }
  178. _, err := db.UserProfileUpdate(eg, userProfile.Uid, userProfile)
  179. if err != nil {
  180. logx.Warn(err)
  181. }
  182. return
  183. }