package consume import ( "applet/app/cfg" utils2 "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/EggPlanet/egg_system_rules.git" "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy" "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/enum" md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md" es2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/utils/es" "code.fnuoos.com/go_rely_warehouse/zyos_go_es.git/es" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "context" "encoding/json" "errors" "fmt" "github.com/olivere/elastic/v7" "github.com/streadway/amqp" "strings" "time" ) func EggCommentDataConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>EggCommentDataConsume>>>>>>>>>>>>") 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 = handleEggCommentDataConsume(res.Body) if err != nil { fmt.Println("EggCommentDataConsume:::::", err.Error()) utils2.FilePutContents("EggCommentDataConsume_ERR", utils2.SerializeStr(map[string]interface{}{ "body": res.Body, "err": err.Error(), })) } //_ = res.Reject(false) err = res.Ack(true) fmt.Println("err ::: ", err) } else { panic(errors.New("error getting message")) } } } func handleEggCommentDataConsume(msgData []byte) error { time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒 // 1.解析mq中queue的数据结构体 var msg *md2.EggSendFriendCircleData err := json.Unmarshal(msgData, &msg) if err != nil { return err } year, week := time.Now().ISOWeek() yearStr := utils2.IntToStr(year) weekStr := utils2.IntToStr(week) index := es2.GetAppointIndexFromAlias(yearStr, weekStr) id := fmt.Sprintf("%d%d_%d", year, week, msg.Uid) script := elastic.NewScript("ctx._source.forum_comments_nums += params.inc").Param("inc", 1) updateDoc, err := es.EsClient.Update(). Index(index). Id(id). Script(script). Do(context.Background()) if err != nil { if strings.Contains(err.Error(), "elastic: Error 404 (Not Found)") { // 蛋蛋分数据还不存在,创建蛋蛋分数据 now := time.Now().Format("2006-01-02 15:04:05") err1 := egg_energy.CreateEsScoreAndAssignValuesDoc(index, id, msg.Uid, enum.ForumCommentsNums, "1", now) if err1 != nil { return err1 } return nil } fmt.Println("EggCommentDataConsumeUpdateDoc_ERR::::", err.Error()) return err } fmt.Println("updateDoc==========>", updateDoc) return nil }