From 4935e2f4b48eccbd2701cbee2c073d5a9fa09d98 Mon Sep 17 00:00:00 2001 From: shenjiachi Date: Mon, 9 Dec 2024 10:22:40 +0800 Subject: [PATCH] update --- consume/egg_canal_violate_nums_consume.go | 3 +- consume/egg_record_active_consume.go | 114 ++++++++++++++++++++++ consume/init.go | 1 + consume/md/consume_key.go | 12 ++- go.mod | 2 +- 5 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 consume/egg_record_active_consume.go diff --git a/consume/egg_canal_violate_nums_consume.go b/consume/egg_canal_violate_nums_consume.go index 243bffc..7bdb3ce 100644 --- a/consume/egg_canal_violate_nums_consume.go +++ b/consume/egg_canal_violate_nums_consume.go @@ -84,13 +84,14 @@ func handleEggCanalViolateNumsConsume(msg []byte) error { } } } + // 3. 监听删除信息 if canalMsg.Type == md2.CanalMsgDeleteSqlType { for _, item := range canalMsg.Data { parentUid := item.ParentUid id := fmt.Sprintf("%d%d-%d", year, week, parentUid) // 减少违规次数记录 - script := elastic.NewScript("ctx._source.violate_nums -= params.inc").Param("inc", 1) + script := elastic.NewScript("ctx._source.violate_nums -= params.dec").Param("dec", 1) _, err = es.EsClient.Update(). Index(index). Id(id). diff --git a/consume/egg_record_active_consume.go b/consume/egg_record_active_consume.go new file mode 100644 index 0000000..2c96b54 --- /dev/null +++ b/consume/egg_record_active_consume.go @@ -0,0 +1,114 @@ +package consume + +import ( + "applet/app/cfg" + "applet/app/db" + utils2 "applet/app/utils" + "applet/app/utils/logx" + "applet/consume/md" + "code.fnuoos.com/EggPlanet/egg_models.git/src/implement" + "code.fnuoos.com/EggPlanet/egg_system_rules.git" + 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" + "time" +) + +func EggRecordActiveDataConsume(queue md.MqQueue) { + fmt.Println(">>>>>>>>>>>>EggRecordActiveDataConsume>>>>>>>>>>>>") + 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 = handleEggRecordActiveDataConsume(res.Body) + if err != nil { + fmt.Println("EggRecordActiveDataConsume_ERR:::::", err.Error()) + utils2.FilePutContents("EggRecordActiveDataConsume_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 handleEggRecordActiveDataConsume(msgData []byte) error { + time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒 + // 1.解析mq中queue的数据结构体 + var msg *md2.EggStructForRecordActiveData + 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) + + // 2.获取用户直推上级 + userRelateDb := implement.NewUserRelateDb(db.Db) + userRelate, err := userRelateDb.GetUserParentUserRelate(msg.Uid) + if err != nil { + return err + } + + // 3.添加团队活跃次数 + if userRelate != nil { + parentUid := userRelate.ParentUid + parentEsId := fmt.Sprintf("%d%d-%d", year, week, parentUid) + // 增加团队活跃次数记录 + script := elastic.NewScript("ctx._source.team_activity_nums += params.inc").Param("inc", 1) + _, err = es.EsClient.Update(). + Index(index). + Id(parentEsId). + Script(script). + Do(context.Background()) + if err != nil { + return err + } + } + + // 4.更新签到次数和个人活跃积分 + esId := fmt.Sprintf("%d%d-%d", year, week, msg.Uid) + script := elastic.NewScript(` + ctx._source.sign_in_nums += params.signInInc; + ctx._source.person_add_activity_value += params.activeValueInc; + `).Param("signInInc", 1).Param("activeValueInc", utils2.StrToInt(msg.TotalPersonEggPoints)) + + _, err = es.EsClient.Update(). + Index(index). + Id(esId). + Script(script). + Do(context.Background()) + if err != nil { + return err + } + return nil +} diff --git a/consume/init.go b/consume/init.go index ff7a468..440872f 100644 --- a/consume/init.go +++ b/consume/init.go @@ -34,6 +34,7 @@ func initConsumes() { jobs[consumeMd.EggCanalInviteUserNumsFunName] = EggCanalInviteUserNumsConsume // 处理拉新人数 jobs[consumeMd.EggCanalViolateNumsFunName] = EggCanalViolateNumsConsume // 处理违规次数 jobs[consumeMd.IMEggEnergySendRedPackageFunName] = IMEggEnergySendRedPackageConsume // 处理用户发送红包次数 + jobs[consumeMd.EggRecordActiveDataFunName] = EggRecordActiveDataConsume // 用户签到后更新活跃数据 } func Run() { diff --git a/consume/md/consume_key.go b/consume/md/consume_key.go index 499765b..2b1a1f7 100644 --- a/consume/md/consume_key.go +++ b/consume/md/consume_key.go @@ -67,7 +67,7 @@ var RabbitMqQueueKeyList = []*MqQueue{ }, { ExchangeName: "egg.app", - Name: "egg_fin_withdraw_apply_error_queue", + Name: "egg_fin_withdraw_apply", Type: DirectQueueType, IsPersistent: false, RoutKey: "egg_fin_withdraw_apply", @@ -182,6 +182,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ BindKey: "", ConsumeFunName: "IMEggEnergySendRedPackageConsume", }, + { + ExchangeName: "egg.app", + Name: "egg_record_active_queue", + Type: DirectQueueType, + IsPersistent: false, + RoutKey: "egg_record_active", + BindKey: "", + ConsumeFunName: "EggRecordActiveDataConsume", + }, } const ( @@ -204,4 +213,5 @@ const ( EggCanalInviteUserNumsFunName = "EggCanalInviteUserNumsConsume" EggCanalViolateNumsFunName = "EggCanalViolateNumsConsume" IMEggEnergySendRedPackageFunName = "IMEggEnergySendRedPackageConsume" + EggRecordActiveDataFunName = "EggRecordActiveDataConsume" ) diff --git a/go.mod b/go.mod index de08ee1..7d3d2e8 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ go 1.19 require ( code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241206115326-8cbc93c7c64d - code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.4-0.20241207095445-64c8aa0b486e + code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.4-0.20241209020046-5e439665e263 code.fnuoos.com/go_rely_warehouse/zyos_go_es.git v1.0.1-0.20241118083738-0f22da9ba0be code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.5 github.com/boombuler/barcode v1.0.1