|
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586 |
- package consume
-
- import (
- "applet/app/cfg"
- "applet/app/db"
- "applet/app/utils"
- "applet/app/utils/logx"
- "applet/consume/md"
- "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
- md2 "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/md"
- honey_dividend "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/honey_dividend"
- "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles"
- "encoding/json"
- "errors"
- "fmt"
- "github.com/streadway/amqp"
- )
-
- func DividendRewardConsume(queue md.MqQueue) {
- fmt.Println(">>>>>>>>>>>>DividendRewardConsume>>>>>>>>>>>>")
- ch, err := rabbit.Cfg.Pool.GetChannel()
- if err != nil {
- logx.Error(err)
- return
- }
- defer ch.Release()
- //1、将自己绑定到交换机上
- ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
- //2、取出数据进行消费
- ch.Qos(1)
- delivery := ch.Consume(queue.Name, false)
-
- one_circles.Init(cfg.RedisAddr)
-
- var res amqp.Delivery
- var ok bool
- for {
- res, ok = <-delivery
- if ok == true {
- err = handleDividendRewardConsume(res.Body)
- if err != nil {
- fmt.Println("err ::: ", err)
- utils.FilePutContents("ServiceAwardDividendRelationConsume_ERR", "[err]:"+err.Error())
- _ = res.Reject(true) //TODO::拒绝 Ack
- } else {
- _ = res.Ack(true)
- }
- } else {
- panic(errors.New("error getting message"))
- }
- }
- fmt.Println("get msg done")
- }
-
- func handleDividendRewardConsume(msgData []byte) error {
- //1、解析mq中queue的数据结构体
- var msg *md2.HoneyDividendForUserDividend
- err := json.Unmarshal(msgData, &msg)
- if err != nil {
- return err
- }
-
- engine := db.DBs[msg.MasterId]
- session := engine.NewSession()
- defer func() {
- session.Close()
- if err := recover(); err != nil {
- _ = logx.Error(err)
- }
- }()
- session.Begin()
-
- err = honey_dividend.DealDividendReward(session, *msg)
- if err != nil {
- _ = session.Rollback()
- fmt.Println("err:::::!!!", err.Error())
- return err
- }
-
- err = session.Commit()
- if err != nil {
- _ = session.Rollback()
- return errors.New("事务提交失败")
- }
- return nil
- }
|