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

jpush_record_consume.go 2.9 KiB

2 weeks ago
1 week ago
1 week ago
1 week ago
1 week ago
1 week ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  1. package consume
  2. import (
  3. "applet/app/cfg"
  4. "applet/app/db"
  5. "code.fnuoos.com/EggPlanet/egg_models.git/src/dao"
  6. utils2 "applet/app/utils"
  7. "applet/app/utils/cache"
  8. "applet/app/utils/logx"
  9. "applet/consume/md"
  10. "code.fnuoos.com/EggPlanet/egg_models.git/src/implement"
  11. "code.fnuoos.com/EggPlanet/egg_models.git/src/model"
  12. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  13. "code.fnuoos.com/EggPlanet/egg_system_rules.git/jPush"
  14. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  15. "encoding/json"
  16. "errors"
  17. "fmt"
  18. "github.com/streadway/amqp"
  19. "strings"
  20. "time"
  21. )
  22. func JpushRecordConsume(queue md.MqQueue) {
  23. fmt.Println(">>>>>>>>>>>>JpushRecordConsume>>>>>>>>>>>>")
  24. ch, err := rabbit.Cfg.Pool.GetChannel()
  25. if err != nil {
  26. logx.Error(err)
  27. return
  28. }
  29. defer ch.Release()
  30. //1、将自己绑定到交换机上
  31. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  32. //2、取出数据进行消费
  33. ch.Qos(1)
  34. delivery := ch.Consume(queue.Name, false)
  35. egg_system_rules.Init(cfg.RedisAddr)
  36. var res amqp.Delivery
  37. var ok bool
  38. for {
  39. res, ok = <-delivery
  40. if ok == true {
  41. err = handleJpushRecordConsume(res.Body)
  42. if err != nil {
  43. fmt.Println("JpushRecordConsume_ERR:::::", err.Error())
  44. utils2.FilePutContents("JpushRecordConsume_ERR", utils2.SerializeStr(map[string]interface{}{
  45. "body": res.Body,
  46. "err": err.Error(),
  47. }))
  48. _ = res.Reject(false)
  49. //TODO::重新推回队列末尾,避免造成队列堵塞
  50. var msg *md.JpushRecordFundData
  51. json.Unmarshal(res.Body, &msg)
  52. msg.Num = "1"
  53. ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
  54. } else {
  55. //_ = res.Reject(false)
  56. err = res.Ack(true)
  57. }
  58. fmt.Println("err ::: ", err)
  59. } else {
  60. panic(errors.New("error getting message"))
  61. }
  62. }
  63. fmt.Println("get msg done")
  64. }
  65. func handleJpushRecordConsume(msgData []byte) error {
  66. time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
  67. // 1.解析mq中queue的数据结构体
  68. var msg *md.JpushRecordFundData
  69. err := json.Unmarshal(msgData, &msg)
  70. if err != nil {
  71. return err
  72. }
  73. engine := db.Db
  74. redisConn := cache.GetPool().Get()
  75. sysCfgDb := implement.NewSysCfgDb(engine, redisConn)
  76. defer func(sysCfgDb dao.SysCfgDao) {
  77. err := sysCfgDb.Close()
  78. if err != nil {
  79. logx.Error("redis close err:" + err.Error()) // 记录错误信息
  80. }
  81. }(sysCfgDb)
  82. jpushKey := sysCfgDb.SysCfgGetWithDb("jpush_key")
  83. jpushSecret := sysCfgDb.SysCfgGetWithDb("jpush_secret")
  84. if msg.Target == "0" { //广播全部
  85. _, err := jPush.PushAllUser(jpushKey, jpushSecret, msg.Title, msg.Content, msg.Platform, nil)
  86. if err != nil && msg.Num != "1" {
  87. return err
  88. }
  89. } else {
  90. _, err = jPush.PushMoreUser(jpushKey, jpushSecret, msg.Title, msg.Content, msg.Platform, strings.Split(msg.UserId, ","), nil)
  91. if err != nil && msg.Num != "1" {
  92. return err
  93. }
  94. }
  95. if msg.Id != "" {
  96. engine.Where("id=?", msg.Id).Cols("state").Update(&model.JpushRecord{State: 1})
  97. }
  98. return nil
  99. }