蛋蛋星球RabbitMq消费项目
Non puoi selezionare più di 25 argomenti Gli argomenti devono iniziare con una lettera o un numero, possono includere trattini ('-') e possono essere lunghi fino a 35 caratteri.
 
 
 
 

79 righe
2.1 KiB

  1. package consume
  2. import (
  3. "applet/app/cfg"
  4. "applet/app/db"
  5. md4 "applet/app/md"
  6. utils2 "applet/app/utils"
  7. "applet/app/utils/logx"
  8. "applet/consume/md"
  9. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  10. "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy"
  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. )
  18. func EggEnergyAutoExchangeGreenEnergyConsume(queue md.MqQueue) {
  19. fmt.Println(">>>>>>>>>>>>EggEnergyAutoExchangeGreenEnergyConsume>>>>>>>>>>>>")
  20. ch, err := rabbit.Cfg.Pool.GetChannel()
  21. if err != nil {
  22. logx.Error(err)
  23. return
  24. }
  25. defer ch.Release()
  26. //1、将自己绑定到交换机上
  27. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  28. //2、取出数据进行消费
  29. ch.Qos(1)
  30. delivery := ch.Consume(queue.Name, false)
  31. egg_system_rules.Init(cfg.RedisAddr)
  32. var res amqp.Delivery
  33. var ok bool
  34. for {
  35. res, ok = <-delivery
  36. if ok == true {
  37. err = handleEggEnergyAutoExchangeGreenEnergyConsume(res.Body, ch)
  38. if err != nil {
  39. fmt.Println("EggEnergyAutoExchangeGreenEnergyConsume_ERR:::::", err.Error())
  40. utils2.FilePutContents("EggEnergyAutoExchangeGreenEnergyConsume_ERR", utils2.SerializeStr(map[string]interface{}{
  41. "body": res.Body,
  42. "err": err.Error(),
  43. }))
  44. _ = res.Reject(true) //TODO::拒绝
  45. } else {
  46. err = res.Ack(true)
  47. fmt.Println("err ::: ", err)
  48. }
  49. } else {
  50. panic(errors.New("error getting message"))
  51. }
  52. }
  53. fmt.Println("get msg done")
  54. }
  55. func handleEggEnergyAutoExchangeGreenEnergyConsume(msgData []byte, ch *rabbit.Channel) error {
  56. //1、解析mq中queue的数据结构体
  57. var msg *md2.EggEnergyStructForAutoExchangeGreenEnergy
  58. err := json.Unmarshal(msgData, &msg)
  59. if err != nil {
  60. return err
  61. }
  62. engine := db.Db
  63. //2、调用制度方法
  64. err = egg_energy.ActivityCoinAutoExchangeEggPersonEnergy(engine, *msg)
  65. if err != nil {
  66. return err
  67. }
  68. //3、给到直推奖励
  69. ch.Publish(md2.EggEnergyExchange, md4.EggEnergyRecruitNewUserRewardMq{
  70. Uid: msg.Uid,
  71. }, "egg_energy_recruit_new_user_reward")
  72. return nil
  73. }