蛋蛋星球RabbitMq消费项目
Nie możesz wybrać więcej, niż 25 tematów Tematy muszą się zaczynać od litery lub cyfry, mogą zawierać myślniki ('-') i mogą mieć do 35 znaków.
 
 
 
 
 
 

70 wiersze
1.9 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. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  9. "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy"
  10. md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md"
  11. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  12. "encoding/json"
  13. "errors"
  14. "fmt"
  15. "github.com/streadway/amqp"
  16. )
  17. func EggEnergyStartExchangeGreenEnergyConsume(queue md.MqQueue) {
  18. fmt.Println(">>>>>>>>>>>>EggEnergyStartExchangeGreenEnergyConsume>>>>>>>>>>>>")
  19. ch, err := rabbit.Cfg.Pool.GetChannel()
  20. if err != nil {
  21. logx.Error(err)
  22. return
  23. }
  24. defer ch.Release()
  25. //1、将自己绑定到交换机上
  26. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  27. //2、取出数据进行消费
  28. ch.Qos(1)
  29. delivery := ch.Consume(queue.Name, false)
  30. egg_system_rules.Init(cfg.RedisAddr)
  31. var res amqp.Delivery
  32. var ok bool
  33. for {
  34. res, ok = <-delivery
  35. if ok == true {
  36. err = handleEggEnergyStartExchangeGreenEnergyConsume(res.Body, ch)
  37. if err != nil {
  38. fmt.Println("EggEnergyStartExchangeGreenEnergyConsume_ERR:::::", err.Error())
  39. utils2.FilePutContents("EggEnergyStartExchangeGreenEnergyConsume_ERR", utils2.SerializeStr(map[string]interface{}{
  40. "body": res.Body,
  41. "err": err.Error(),
  42. }))
  43. }
  44. _ = res.Reject(true) //TODO::拒绝
  45. //err = res.Ack(true)
  46. fmt.Println("err ::: ", err)
  47. } else {
  48. panic(errors.New("error getting message"))
  49. }
  50. }
  51. fmt.Println("get msg done")
  52. }
  53. func handleEggEnergyStartExchangeGreenEnergyConsume(msgData []byte, ch *rabbit.Channel) error {
  54. //1、解析mq中queue的数据结构体
  55. var msg *md2.EggEnergyStructForStartExchangeGreenEnergy
  56. err := json.Unmarshal(msgData, &msg)
  57. if err != nil {
  58. return err
  59. }
  60. engine := db.Db
  61. //2、调用制度方法
  62. err = egg_energy.ActivityCoinStartExchangeEggEnergy(engine, *msg, ch)
  63. return err
  64. }