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

93 satır
2.6 KiB

  1. package consume
  2. import (
  3. utils2 "applet/app/utils"
  4. "applet/app/utils/logx"
  5. "applet/consume/md"
  6. md2 "applet/es/md"
  7. es2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/utils/es"
  8. "code.fnuoos.com/go_rely_warehouse/zyos_go_es.git/es"
  9. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  10. "context"
  11. "encoding/json"
  12. "errors"
  13. "fmt"
  14. "github.com/olivere/elastic/v7"
  15. "github.com/streadway/amqp"
  16. "time"
  17. )
  18. func EggCanalInviteUserNumsConsume(queue md.MqQueue) {
  19. fmt.Println(">>>>>>>>>>>EggCanalInviteUserNumsConsume>>>>>>>>>>>>>")
  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(1000)
  30. delivery := ch.Consume(queue.Name, true) //设置自动应答
  31. var res amqp.Delivery
  32. var ok bool
  33. for {
  34. res, ok = <-delivery
  35. if ok == true {
  36. //fmt.Println(string(res.Body))
  37. fmt.Println(">>>>>>>>>>>>>>>>>>EggCanalInviteUserNumsConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
  38. err = handleEggCanalInviteUserNumsConsume(res.Body)
  39. if err != nil {
  40. fmt.Println("EggCanalInviteUserNumsConsume_ERR:::::", err.Error())
  41. utils2.FilePutContents("EggCanalInviteUserNumsConsume_ERR", utils2.SerializeStr(map[string]interface{}{
  42. "body": res.Body,
  43. "err": err.Error(),
  44. }))
  45. }
  46. //_ = res.Reject(false)
  47. //_ = res.Ack(true)
  48. } else {
  49. panic(errors.New("error getting message"))
  50. }
  51. }
  52. }
  53. func handleEggCanalInviteUserNumsConsume(msg []byte) error {
  54. //1、解析canal采集至mq中queue的数据结构体
  55. var canalMsg *md.CanalUserRelateMessage[md.CanalUserRelate]
  56. err := json.Unmarshal(msg, &canalMsg)
  57. if err != nil {
  58. fmt.Println("EggCanalInviteUserNumsConsumeFaliedUnMarshal_ERR:::::", err.Error())
  59. return nil
  60. }
  61. year, week := time.Now().ISOWeek()
  62. yearStr := utils2.IntToStr(year)
  63. weekStr := utils2.IntToStr(week)
  64. index := es2.GetAppointIndexFromAlias(yearStr, weekStr)
  65. // 2. 监听插入信息
  66. if canalMsg.Type == md2.CanalMsgInsertSqlType {
  67. for _, item := range canalMsg.Data {
  68. parentUid := item.ParentUid
  69. id := fmt.Sprintf("%d%d_%s", year, week, parentUid)
  70. // 新增拉新人数
  71. script := elastic.NewScript("ctx._source.invite_user_nums += params.inc").Param("inc", 1)
  72. service, err := es.EsClient.Update().
  73. Index(index).
  74. Id(id).
  75. Script(script).
  76. Do(context.Background())
  77. if err != nil {
  78. fmt.Println("EggCanalInviteUserNumsConsumeFailedUpdateInviteUserNums_ERR:::::", err.Error())
  79. return err
  80. }
  81. fmt.Println(service)
  82. }
  83. }
  84. fmt.Println("EggCanalInviteUserNumsConsume_Succeed::::")
  85. return nil
  86. }