From 4935e2f4b48eccbd2701cbee2c073d5a9fa09d98 Mon Sep 17 00:00:00 2001 From: shenjiachi Date: Mon, 9 Dec 2024 10:22:40 +0800 Subject: [PATCH 1/2] 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 From 9dbf9cdf6f0b0bb4c7c3a483783193febd23bee0 Mon Sep 17 00:00:00 2001 From: shenjiachi Date: Mon, 9 Dec 2024 14:45:55 +0800 Subject: [PATCH 2/2] add record active data --- ...canal_person_add_activity_value_consume.go | 102 ++++++++++++++++++ consume/egg_canal_violate_nums_consume.go | 41 +++++-- consume/egg_record_active_consume.go | 6 +- .../im_egg_energy_send_red_package_consume.go | 2 +- consume/init.go | 13 +-- consume/md/consume_key.go | 10 ++ ...canal_person_add_activity_value_consume.go | 27 +++++ ...o => md_egg_canal_violate_nums_consume.go} | 0 go.mod | 2 +- 9 files changed, 185 insertions(+), 18 deletions(-) create mode 100644 consume/egg_canal_person_add_activity_value_consume.go create mode 100644 consume/md/md_egg_canal_person_add_activity_value_consume.go rename consume/md/{egg_canal_violate_nums_consume.go => md_egg_canal_violate_nums_consume.go} (100%) diff --git a/consume/egg_canal_person_add_activity_value_consume.go b/consume/egg_canal_person_add_activity_value_consume.go new file mode 100644 index 0000000..90c5c63 --- /dev/null +++ b/consume/egg_canal_person_add_activity_value_consume.go @@ -0,0 +1,102 @@ +package consume + +import ( + "applet/app/db" + utils2 "applet/app/utils" + "applet/app/utils/logx" + "applet/consume/md" + md2 "applet/es/md" + "code.fnuoos.com/EggPlanet/egg_models.git/src/implement" + 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" +) + +// EggCanalPersonAddActivityValueConsume 用户获得活跃积分时更新到 es +func EggCanalPersonAddActivityValueConsume(queue md.MqQueue) { + fmt.Println(">>>>>>>>>>>EggCanalPersonAddActivityValueConsume>>>>>>>>>>>>>") + 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(1000) + delivery := ch.Consume(queue.Name, true) //设置自动应答 + + var res amqp.Delivery + var ok bool + for { + res, ok = <-delivery + if ok == true { + //fmt.Println(string(res.Body)) + fmt.Println(">>>>>>>>>>>>>>>>>>EggCanalPersonAddActivityValueConsume<<<<<<<<<<<<<<<<<<<<<<<<<") + err = handleEggCanalPersonAddActivityValueConsume(res.Body) + if err != nil { + fmt.Println("EggCanalPersonAddActivityValueConsume_ERR:::::", err.Error()) + utils2.FilePutContents("EggCanalPersonAddActivityValueConsume_ERR", utils2.SerializeStr(map[string]interface{}{ + "body": res.Body, + "err": err.Error(), + })) + } + //_ = res.Reject(false) + //_ = res.Ack(true) + } else { + panic(errors.New("error getting message")) + } + } +} + +func handleEggCanalPersonAddActivityValueConsume(msg []byte) error { + //1.解析canal采集至mq中queue的数据结构体 + var canalMsg *md.CanalUserVirtualCoinFlowMessage[md.CanalUserVirtualCoinFlow] + err := json.Unmarshal(msg, &canalMsg) + if err != nil { + return nil + } + + year, week := time.Now().ISOWeek() + yearStr := utils2.IntToStr(year) + weekStr := utils2.IntToStr(week) + index := es2.GetAppointIndexFromAlias(yearStr, weekStr) + + settingDb := implement.NewEggEnergyBasicSettingDb(db.Db) + setting, err := settingDb.EggEnergyBasicSettingGetOne() + if err != nil { + return err + } + personEggPointsCoinId := setting.PersonEggPointsCoinId + // 2. 监听插入信息 + if canalMsg.Type == md2.CanalMsgInsertSqlType { + for _, item := range canalMsg.Data { + uid := item.Uid + id := fmt.Sprintf("%d%d-%d", year, week, uid) + + if item.CoinId != personEggPointsCoinId && item.Direction != 1 { + continue + } + + // 3. 增加个人活跃积分 + script := elastic.NewScript("ctx._source.person_add_activity_value += params.inc").Param("inc", item.Amount) + _, err = es.EsClient.Update(). + Index(index). + Id(id). + Script(script). + Do(context.Background()) + if err != nil { + return err + } + } + } + return nil +} diff --git a/consume/egg_canal_violate_nums_consume.go b/consume/egg_canal_violate_nums_consume.go index 7bdb3ce..cb42785 100644 --- a/consume/egg_canal_violate_nums_consume.go +++ b/consume/egg_canal_violate_nums_consume.go @@ -1,10 +1,12 @@ package consume import ( + "applet/app/db" utils2 "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" md2 "applet/es/md" + "code.fnuoos.com/EggPlanet/egg_models.git/src/implement" 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" @@ -17,6 +19,7 @@ import ( "time" ) +// EggCanalViolateNumsConsume 更新违规次数 func EggCanalViolateNumsConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>EggCanalViolateNumsConsume>>>>>>>>>>>>>") ch, err := rabbit.Cfg.Pool.GetChannel() @@ -56,7 +59,7 @@ func EggCanalViolateNumsConsume(queue md.MqQueue) { func handleEggCanalViolateNumsConsume(msg []byte) error { //1、解析canal采集至mq中queue的数据结构体 - var canalMsg *md.CanalUserRelateMessage[md.CanalUserRelate] + var canalMsg *md.CanalTagRecordsMessage[md.CanalTagRecords] err := json.Unmarshal(msg, &canalMsg) if err != nil { return nil @@ -68,11 +71,23 @@ func handleEggCanalViolateNumsConsume(msg []byte) error { index := es2.GetAppointIndexFromAlias(yearStr, weekStr) // 2. 监听插入信息 if canalMsg.Type == md2.CanalMsgInsertSqlType { + tagDb := implement.NewUserTagDb(db.Db) for _, item := range canalMsg.Data { - parentUid := item.ParentUid - id := fmt.Sprintf("%d%d-%d", year, week, parentUid) + tag, err1 := tagDb.UserTagGetOneByParams(map[string]interface{}{ + "key": "id", + "value": item.TagId, + }) + if err1 != nil { + return err1 + } + // 2.1. 判断是否为处罚标签 + if tag.IsPunish == 0 { + continue + } + uid := item.Uid + id := fmt.Sprintf("%d%d-%d", year, week, uid) - // 增加违规次数记录 + // 2.2. 增加违规次数记录 script := elastic.NewScript("ctx._source.violate_nums += params.inc").Param("inc", 1) _, err = es.EsClient.Update(). Index(index). @@ -86,11 +101,23 @@ func handleEggCanalViolateNumsConsume(msg []byte) error { } // 3. 监听删除信息 if canalMsg.Type == md2.CanalMsgDeleteSqlType { + tagDb := implement.NewUserTagDb(db.Db) for _, item := range canalMsg.Data { - parentUid := item.ParentUid - id := fmt.Sprintf("%d%d-%d", year, week, parentUid) + tag, err1 := tagDb.UserTagGetOneByParams(map[string]interface{}{ + "key": "id", + "value": item.TagId, + }) + if err1 != nil { + return err1 + } + // 3.1 判断是否为处罚标签 + if tag.IsPunish == 0 { + continue + } + uid := item.Uid + id := fmt.Sprintf("%d%d-%d", year, week, uid) - // 减少违规次数记录 + // 3.2 减少违规次数记录 script := elastic.NewScript("ctx._source.violate_nums -= params.dec").Param("dec", 1) _, err = es.EsClient.Update(). Index(index). diff --git a/consume/egg_record_active_consume.go b/consume/egg_record_active_consume.go index 2c96b54..1b662f4 100644 --- a/consume/egg_record_active_consume.go +++ b/consume/egg_record_active_consume.go @@ -21,6 +21,7 @@ import ( "time" ) +// EggRecordActiveDataConsume 更新团队活跃次数和签到次数 func EggRecordActiveDataConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>EggRecordActiveDataConsume>>>>>>>>>>>>") ch, err := rabbit.Cfg.Pool.GetChannel() @@ -95,12 +96,11 @@ func handleEggRecordActiveDataConsume(msgData []byte) error { } } - // 4.更新签到次数和个人活跃积分 + // 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)) + `).Param("signInInc", 1) _, err = es.EsClient.Update(). Index(index). diff --git a/consume/im_egg_energy_send_red_package_consume.go b/consume/im_egg_energy_send_red_package_consume.go index 1f6ffbe..32c05ad 100644 --- a/consume/im_egg_energy_send_red_package_consume.go +++ b/consume/im_egg_energy_send_red_package_consume.go @@ -70,7 +70,7 @@ func handleIMEggEnergySendRedPackageConsume(msgData []byte) error { weekStr := utils2.IntToStr(week) index := es2.GetAppointIndexFromAlias(yearStr, weekStr) id := fmt.Sprintf("%d%d-%d", year, week, msg.Uid) - // 新增拉新人数 + // 3.新增发红包次数 script := elastic.NewScript("ctx._source.send_red_package_nums += params.inc").Param("inc", 1) _, err = es.EsClient.Update(). Index(index). diff --git a/consume/init.go b/consume/init.go index 440872f..f818864 100644 --- a/consume/init.go +++ b/consume/init.go @@ -29,12 +29,13 @@ func initConsumes() { jobs[consumeMd.EggEnergySettlementPublicGiveActivityCoinFunName] = EggEnergySettlementPublicGiveActivityCoinConsume jobs[consumeMd.EggEnergyStartExchangeGreenEnergyFunName] = EggEnergyStartExchangeGreenEnergyConsume jobs[consumeMd.EggEnergyAutoExchangeGreenEnergyFunName] = EggEnergyAutoExchangeGreenEnergyConsume - jobs[consumeMd.EggEnergyNewUserRegisterDataFunName] = EggEnergyNewUserRegisterDataConsume // 新用户注册 - jobs[consumeMd.EggEnergyDealUserECPMFunName] = EggEnergyDealUserECPMConsume // 处理给用户Ecpm值 - jobs[consumeMd.EggCanalInviteUserNumsFunName] = EggCanalInviteUserNumsConsume // 处理拉新人数 - jobs[consumeMd.EggCanalViolateNumsFunName] = EggCanalViolateNumsConsume // 处理违规次数 - jobs[consumeMd.IMEggEnergySendRedPackageFunName] = IMEggEnergySendRedPackageConsume // 处理用户发送红包次数 - jobs[consumeMd.EggRecordActiveDataFunName] = EggRecordActiveDataConsume // 用户签到后更新活跃数据 + jobs[consumeMd.EggEnergyNewUserRegisterDataFunName] = EggEnergyNewUserRegisterDataConsume // 新用户注册 + jobs[consumeMd.EggEnergyDealUserECPMFunName] = EggEnergyDealUserECPMConsume // 处理给用户Ecpm值 + jobs[consumeMd.EggCanalInviteUserNumsFunName] = EggCanalInviteUserNumsConsume // 处理拉新人数 + jobs[consumeMd.EggCanalViolateNumsFunName] = EggCanalViolateNumsConsume // 处理违规次数 + jobs[consumeMd.IMEggEnergySendRedPackageFunName] = IMEggEnergySendRedPackageConsume // 处理用户发送红包次数 + jobs[consumeMd.EggRecordActiveDataFunName] = EggRecordActiveDataConsume // 用户签到后更新活跃数据 + jobs[consumeMd.EggCanalPersonAddActivityValueFunName] = EggCanalPersonAddActivityValueConsume // 用户活跃积分变更时更新es } func Run() { diff --git a/consume/md/consume_key.go b/consume/md/consume_key.go index 2b1a1f7..7266e0c 100644 --- a/consume/md/consume_key.go +++ b/consume/md/consume_key.go @@ -191,6 +191,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ BindKey: "", ConsumeFunName: "EggRecordActiveDataConsume", }, + { + ExchangeName: "egg.canal.topic", + Name: "egg_canal_person_add_activity_value_queue", + Type: DirectQueueType, + IsPersistent: false, + RoutKey: "egg_canal_person_add_activity_value", + BindKey: "", + ConsumeFunName: "EggCanalPersonAddActivityValueConsume", + }, } const ( @@ -214,4 +223,5 @@ const ( EggCanalViolateNumsFunName = "EggCanalViolateNumsConsume" IMEggEnergySendRedPackageFunName = "IMEggEnergySendRedPackageConsume" EggRecordActiveDataFunName = "EggRecordActiveDataConsume" + EggCanalPersonAddActivityValueFunName = "EggCanalPersonAddActivityValueConsume" ) diff --git a/consume/md/md_egg_canal_person_add_activity_value_consume.go b/consume/md/md_egg_canal_person_add_activity_value_consume.go new file mode 100644 index 0000000..09e09e9 --- /dev/null +++ b/consume/md/md_egg_canal_person_add_activity_value_consume.go @@ -0,0 +1,27 @@ +package md + +type CanalUserVirtualCoinFlow struct { + Id int64 `json:"id"` + Uid int64 `json:"uid"` + CoinId int `json:"coin_id"` + Direction int `json:"direction"` + Title string `json:"title"` + Amount string `json:"amount"` + BeforeAmount string `json:"before_amount"` + AfterAmount string `json:"after_amount"` + SysFee string `json:"sys_fee"` + TransferType int `json:"transfer_type"` +} + +type CanalUserVirtualCoinFlowMessage[T any] struct { + Data []T `json:"data"` + Database string `json:"database"` + ES int64 `json:"es"` + ID int64 `json:"id"` + IsDdl bool `json:"isDdl"` + Old []T `json:"old"` + PkNames []string `json:"pkNames"` + Table string `json:"table"` + TS int64 `json:"ts"` + Type string `json:"type"` +} diff --git a/consume/md/egg_canal_violate_nums_consume.go b/consume/md/md_egg_canal_violate_nums_consume.go similarity index 100% rename from consume/md/egg_canal_violate_nums_consume.go rename to consume/md/md_egg_canal_violate_nums_consume.go diff --git a/go.mod b/go.mod index 7d3d2e8..d3b12d7 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ go 1.19 // replace code.fnuoos.com/EggPlanet/egg_system_rules.git => E:/company/Egg/egg_system_rules require ( - code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241206115326-8cbc93c7c64d + code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241209061118-ae1be0db6a70 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