From 08fb0a00a29c3e0a023130f14e53b19ee4fe63c6 Mon Sep 17 00:00:00 2001 From: DengBiao <2319963317@qq.com> Date: Wed, 28 Feb 2024 18:48:16 +0800 Subject: [PATCH 1/2] update --- app/db/dbs_map.go | 4 +- consume/init.go | 11 +++-- consume/md/consume_key.go | 14 +++++- consume/md/md_one_circles.go | 7 +++ consume/one_circles_sign_in_consume.go | 66 ++++++++++++++++++++++++++ etc/cfg.yml | 6 +-- go.mod | 2 +- 7 files changed, 98 insertions(+), 12 deletions(-) create mode 100644 consume/md/md_one_circles.go create mode 100644 consume/one_circles_sign_in_consume.go diff --git a/app/db/dbs_map.go b/app/db/dbs_map.go index 693d4ca..5acfe9d 100644 --- a/app/db/dbs_map.go +++ b/app/db/dbs_map.go @@ -21,7 +21,7 @@ func InitMapDbs(c *cfg.DBCfg, prd bool) { logx.Fatalf("db_mapping not exists : %v", err) } // tables := MapAllDatabases(debug) - if true { + if cfg.Prd { tables = GetAllDatabasePrd() //debug 获取生产 } else { tables = GetAllDatabaseDev() //debug 获取开发 @@ -110,7 +110,7 @@ func GetAllDatabaseDev() *[]model.DbMapping { fmt.Println("cfg.Local is: ", cfg.Local) if cfg.Local { // 本地调试 加快速度 fmt.Println("notice:LOCAL TEST, only masterId:** 99813608 ** available!") - err = Db.Where("deleted_at != ? AND db_master_id=?", 1, 32053480).Find(&m) + err = Db.Where("deleted_at != ? AND db_master_id=?", 1, 31585332).Find(&m) } else { err = Db.Where("deleted_at != ? AND is_dev = '1' ", 1).Find(&m) } diff --git a/consume/init.go b/consume/init.go index 79f5e7f..7d9e233 100644 --- a/consume/init.go +++ b/consume/init.go @@ -70,10 +70,13 @@ func initConsumes() { //jobs[consumeMd.MallAddSupplyGoodsFunName] = MallAddSupplyGoodsConsume //////////////////////////////////////// bigData ///////////////////////////////////////////////////// - jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume - jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume - jobs[consumeMd.ZhiOsUserVisitIpAddressConsumeFunName] = ZhiOsUserVisitIpAddressConsume - jobs[consumeMd.CanalUserVirtualCcoinFlowFunName] = CanalUserVirtualCoinFlowConsume + //jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume + //jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume + //jobs[consumeMd.ZhiOsUserVisitIpAddressConsumeFunName] = ZhiOsUserVisitIpAddressConsume + //jobs[consumeMd.CanalUserVirtualCcoinFlowFunName] = CanalUserVirtualCoinFlowConsume + + //////////////////////////////////////// oneCircles ///////////////////////////////////////////////////// + jobs[consumeMd.OneCirclesSignInGreenEnergyFunName] = OneCirclesSignInGreenEnergyConsume } func Run() { diff --git a/consume/md/consume_key.go b/consume/md/consume_key.go index 8a1be54..b446bac 100644 --- a/consume/md/consume_key.go +++ b/consume/md/consume_key.go @@ -387,7 +387,16 @@ var RabbitMqQueueKeyList = []*MqQueue{ IsPersistent: false, RoutKey: "canal_user_virtual_coin_flow", BindKey: "", - ConsumeFunName: "CanalUserVirtualCcoinFlowConsume", + ConsumeFunName: "CanalUserVirtualCoinFlowConsume", + }, + { + ExchangeName: "one.circles", + Name: "one_circles_sign_in_green_energy", + Type: TopicQueueType, + IsPersistent: false, + RoutKey: "sign_in", + BindKey: "", + ConsumeFunName: "OneCirclesSignInGreenEnergyConsume", }, } @@ -433,5 +442,6 @@ const ( ZhiosAcquisitionConditionDevFunName = "ZhiosAcquisitionConditionDev" SupplyCloudChainFenxiaoNewChangeFunName = "SupplyCloudChainFenxiaoNewChangeConsume" MallAddSupplyGoodsFunName = "MallAddSupplyGoodsConsume" - CanalUserVirtualCcoinFlowFunName = "CanalUserVirtualCcoinFlowConsume" + CanalUserVirtualCcoinFlowFunName = "CanalUserVirtualCoinFlowConsume" + OneCirclesSignInGreenEnergyFunName = "OneCirclesSignInGreenEnergyConsume" ) diff --git a/consume/md/md_one_circles.go b/consume/md/md_one_circles.go new file mode 100644 index 0000000..fa012c5 --- /dev/null +++ b/consume/md/md_one_circles.go @@ -0,0 +1,7 @@ +package md + +type OneCirclesStructForSignIn struct { + MasterId string `json:"master_id"` + Uid int `json:"uid"` + Id int64 `json:"id"` +} diff --git a/consume/one_circles_sign_in_consume.go b/consume/one_circles_sign_in_consume.go new file mode 100644 index 0000000..c0af23c --- /dev/null +++ b/consume/one_circles_sign_in_consume.go @@ -0,0 +1,66 @@ +package consume + +import ( + "applet/app/cfg" + "applet/app/db" + "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 OneCirclesSignInGreenEnergyConsume(queue md.MqQueue) { + fmt.Println(">>>>>>>>>>>>OneCirclesSignInGreenEnergyConsume>>>>>>>>>>>>") + 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) + + one_circles.Init(cfg.RedisAddr) + + var res amqp.Delivery + var ok bool + for { + res, ok = <-delivery + if ok == true { + err = handleOneCirclesSignInGreenEnergy(res.Body) + if err != nil { + fmt.Println("OneCirclesSignInGreenEnergyConsume_ERR:::::", err.Error()) + } + //_ = res.Reject(false) + err = res.Ack(true) + fmt.Println("err ::: ", err) + } else { + panic(errors.New("error getting message")) + } + } + fmt.Println("get msg done") +} + +func handleOneCirclesSignInGreenEnergy(msgData []byte) error { + //1、解析mq中queue的数据结构体 + var msg *md.OneCirclesStructForSignIn + err := json.Unmarshal(msgData, &msg) + if err != nil { + return err + } + + engine := db.DBs[msg.MasterId] + err = one_circles.HandleSettlementSignInGreenEnergy(engine, msg.MasterId, msg.Id, msg.Uid) + fmt.Println("err::::", err) + if err != nil { + return err + } + return nil +} diff --git a/etc/cfg.yml b/etc/cfg.yml index 9bee1ce..cdeed99 100644 --- a/etc/cfg.yml +++ b/etc/cfg.yml @@ -29,10 +29,10 @@ wxapplet_filepath: # 连接官网数据库获取db mapping db: - host: 'zhios123.rwlb.rds.aliyuncs.com:3306' + host: '119.23.182.117:3306' name: 'zyos_website' - user: 'canal' - psw: 'canal' + user: 'root' + psw: 'Fnuo123com@' show_log: true max_lifetime: 30 max_open_conns: 100 diff --git a/go.mod b/go.mod index 3e1ddf9..8e3807e 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( code.fnuoos.com/go_rely_warehouse/zyos_go_condition_statistics.git v1.1.2-0.20240222023917-c31b53f7e8cb code.fnuoos.com/go_rely_warehouse/zyos_go_es.git v1.0.0 code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.4 - code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240119104238-05c3962029ff + code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240228103137-25bbc2fa494f code.fnuoos.com/go_rely_warehouse/zyos_go_pay.git v1.6.2-0.20231116085701-9ba6e19f877b code.fnuoos.com/go_rely_warehouse/zyos_go_third_party_api.git v1.1.21-0.20240126015516-38ca248db2fd github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5 From 9ba676ff623f69997e5ebf96cfa48f2a792a80ba Mon Sep 17 00:00:00 2001 From: DengBiao <2319963317@qq.com> Date: Thu, 29 Feb 2024 15:34:21 +0800 Subject: [PATCH 2/2] update --- consume/init.go | 1 + consume/md/consume_key.go | 10 ++++ consume/one_circles_sign_in_copy_consume.go | 66 +++++++++++++++++++++ go.mod | 2 +- 4 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 consume/one_circles_sign_in_copy_consume.go diff --git a/consume/init.go b/consume/init.go index 7d9e233..0dcca16 100644 --- a/consume/init.go +++ b/consume/init.go @@ -77,6 +77,7 @@ func initConsumes() { //////////////////////////////////////// oneCircles ///////////////////////////////////////////////////// jobs[consumeMd.OneCirclesSignInGreenEnergyFunName] = OneCirclesSignInGreenEnergyConsume + //jobs[consumeMd.OneCirclesSignInCopyGreenEnergyFunName] = OneCirclesSignInCopyGreenEnergyConsume } func Run() { diff --git a/consume/md/consume_key.go b/consume/md/consume_key.go index b446bac..de76af3 100644 --- a/consume/md/consume_key.go +++ b/consume/md/consume_key.go @@ -398,6 +398,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ BindKey: "", ConsumeFunName: "OneCirclesSignInGreenEnergyConsume", }, + { + ExchangeName: "one.circles", + Name: "one_circles_sign_in_green_energy_copy", + Type: TopicQueueType, + IsPersistent: false, + RoutKey: "sign_in_copy", + BindKey: "", + ConsumeFunName: "OneCirclesSignInCopyGreenEnergyConsume", + }, } const ( @@ -444,4 +453,5 @@ const ( MallAddSupplyGoodsFunName = "MallAddSupplyGoodsConsume" CanalUserVirtualCcoinFlowFunName = "CanalUserVirtualCoinFlowConsume" OneCirclesSignInGreenEnergyFunName = "OneCirclesSignInGreenEnergyConsume" + OneCirclesSignInCopyGreenEnergyFunName = "OneCirclesSignInCopyGreenEnergyConsume" ) diff --git a/consume/one_circles_sign_in_copy_consume.go b/consume/one_circles_sign_in_copy_consume.go new file mode 100644 index 0000000..ca458a1 --- /dev/null +++ b/consume/one_circles_sign_in_copy_consume.go @@ -0,0 +1,66 @@ +package consume + +import ( + "applet/app/cfg" + "applet/app/db" + "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 OneCirclesSignInCopyGreenEnergyConsume(queue md.MqQueue) { + fmt.Println(">>>>>>>>>>>>OneCirclesSignInCopyGreenEnergyConsume>>>>>>>>>>>>") + 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) + + one_circles.Init(cfg.RedisAddr) + + var res amqp.Delivery + var ok bool + for { + res, ok = <-delivery + if ok == true { + err = handleOneCirclesSignInCopyGreenEnergy(res.Body) + if err != nil { + fmt.Println("OneCirclesSignInCopyGreenEnergyConsume_ERR:::::", err.Error()) + } + //_ = res.Reject(false) + err = res.Ack(true) + fmt.Println("err ::: ", err) + } else { + panic(errors.New("error getting message")) + } + } + fmt.Println("get msg done") +} + +func handleOneCirclesSignInCopyGreenEnergy(msgData []byte) error { + //1、解析mq中queue的数据结构体 + var msg *md.OneCirclesStructForSignIn + err := json.Unmarshal(msgData, &msg) + if err != nil { + return err + } + + engine := db.DBs[msg.MasterId] + err = one_circles.HandleSettlementSignInGreenEnergy(engine, msg.MasterId, msg.Id, msg.Uid) + fmt.Println("err::::", err) + if err != nil { + return err + } + return nil +} diff --git a/go.mod b/go.mod index 8e3807e..f020309 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( code.fnuoos.com/go_rely_warehouse/zyos_go_condition_statistics.git v1.1.2-0.20240222023917-c31b53f7e8cb code.fnuoos.com/go_rely_warehouse/zyos_go_es.git v1.0.0 code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.4 - code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240228103137-25bbc2fa494f + code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240229073153-b243a6bc836c code.fnuoos.com/go_rely_warehouse/zyos_go_pay.git v1.6.2-0.20231116085701-9ba6e19f877b code.fnuoos.com/go_rely_warehouse/zyos_go_third_party_api.git v1.1.21-0.20240126015516-38ca248db2fd github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5