package consume import ( svc "applet/app/admin/svc/enterprise_manage" "applet/app/cfg" "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" ) func JudgePackageOrdStateQueueConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>JudgePackageOrdStateQueueConsume>>>>>>>>>>>>") 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(5) 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 = handleJudgePackageOrdStateQueueConsume(ch, res.Body) if err != nil { fmt.Println("err ::: ", err) utils.FilePutContents("JudgePackageOrdStateQueueConsume", "[err]:"+err.Error()) //_ = res.Reject(true) //TODO::拒绝 Ack _ = res.Reject(false) } else { _ = res.Ack(true) } } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleJudgePackageOrdStateQueueConsume(ch *rabbit.Channel, msgData []byte) error { //1、解析mq中queue的数据结构体 var msg *md.JudgePackageOrdOrdState err := json.Unmarshal(msgData, &msg) if err != nil { return err } err = svc.DealJudgePackageOrdOrdState(msg.OrdNo) fmt.Println("err::::", err) if err != nil { return err } return nil }