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

honey_dividends_consume.go 2.0 KiB

3 ay önce
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
  1. package consume
  2. import (
  3. "applet/app/cfg"
  4. "applet/app/db"
  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. md2 "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/md"
  10. honey_dividend "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/honey_dividend"
  11. "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles"
  12. "encoding/json"
  13. "errors"
  14. "fmt"
  15. "github.com/streadway/amqp"
  16. )
  17. func DividendRewardConsume(queue md.MqQueue) {
  18. fmt.Println(">>>>>>>>>>>>DividendRewardConsume>>>>>>>>>>>>")
  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. one_circles.Init(cfg.RedisAddr)
  31. var res amqp.Delivery
  32. var ok bool
  33. for {
  34. res, ok = <-delivery
  35. if ok == true {
  36. err = handleDividendRewardConsume(res.Body)
  37. if err != nil {
  38. fmt.Println("err ::: ", err)
  39. utils.FilePutContents("ServiceAwardDividendRelationConsume_ERR", "[err]:"+err.Error())
  40. _ = res.Reject(true) //TODO::拒绝 Ack
  41. } else {
  42. _ = res.Ack(true)
  43. }
  44. } else {
  45. panic(errors.New("error getting message"))
  46. }
  47. }
  48. fmt.Println("get msg done")
  49. }
  50. func handleDividendRewardConsume(msgData []byte) error {
  51. //1、解析mq中queue的数据结构体
  52. var msg *md2.HoneyDividendForUserDividend
  53. err := json.Unmarshal(msgData, &msg)
  54. if err != nil {
  55. return err
  56. }
  57. engine := db.DBs[msg.MasterId]
  58. session := engine.NewSession()
  59. defer func() {
  60. session.Close()
  61. if err := recover(); err != nil {
  62. _ = logx.Error(err)
  63. }
  64. }()
  65. session.Begin()
  66. err = honey_dividend.DealDividendReward(session, *msg)
  67. if err != nil {
  68. _ = session.Rollback()
  69. fmt.Println("err:::::!!!", err.Error())
  70. return err
  71. }
  72. err = session.Commit()
  73. if err != nil {
  74. _ = session.Rollback()
  75. return errors.New("事务提交失败")
  76. }
  77. return nil
  78. }