package consume import ( "applet/app/cfg" "applet/app/db" "applet/app/svc/sys_cfg" utils2 "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/EggPlanet/egg_models.git/src/model" "code.fnuoos.com/EggPlanet/egg_system_rules.git" "code.fnuoos.com/EggPlanet/egg_system_rules.git/jPush" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" "strings" "time" ) func JpushRecordConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>JpushRecordConsume>>>>>>>>>>>>") 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) egg_system_rules.Init(cfg.RedisAddr) var res amqp.Delivery var ok bool for { res, ok = <-delivery if ok == true { err = handleJpushRecordConsume(res.Body) if err != nil { fmt.Println("JpushRecordConsume_ERR:::::", err.Error()) utils2.FilePutContents("JpushRecordConsume_ERR", utils2.SerializeStr(map[string]interface{}{ "body": res.Body, "err": err.Error(), })) _ = res.Reject(false) //TODO::重新推回队列末尾,避免造成队列堵塞 var msg *md.JpushRecordFundData json.Unmarshal(res.Body, &msg) msg.Num = "1" ch.Publish(queue.ExchangeName, msg, queue.RoutKey) } else { //_ = res.Reject(false) err = res.Ack(true) } fmt.Println("err ::: ", err) } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleJpushRecordConsume(msgData []byte) error { time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒 // 1.解析mq中queue的数据结构体 var msg *md.JpushRecordFundData err := json.Unmarshal(msgData, &msg) if err != nil { return err } engine := db.Db sysCfgDb := sys_cfg.NewSysCfgDb(engine) jpushKey := sysCfgDb.SysCfgGetWithDb("jpush_key") jpushSecret := sysCfgDb.SysCfgGetWithDb("jpush_secret") if msg.Target == "0" { //广播全部 _, err := jPush.PushAllUser(jpushKey, jpushSecret, msg.Title, msg.Content, msg.Platform, nil) if err != nil && msg.Num != "1" { return err } } else { _, err = jPush.PushMoreUser(jpushKey, jpushSecret, msg.Title, msg.Content, msg.Platform, strings.Split(msg.UserId, ","), nil) if err != nil && msg.Num != "1" { return err } } if msg.Id != "" { engine.Where("id=?", msg.Id).Cols("state").Update(&model.JpushRecord{State: 1}) } return nil }