go_rely_warehouse
/
zyos_go_mq_consume
4
0
Fork 0
Code Issues 0 Pull Requests 0 Releases 0 Wiki Activity
Browse Source

Merge remote-tracking branch 'origin/master'

master
huangjiajun 3 weeks ago
parent
c876f724d3 d242340ecf
commit
73da568a6d
6 changed files with 191 additions and 52 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -1
      app/db/dbs_map.go
  2. +101
    -0
      consume/canal_service_award_dividend_relation_consume.go
  3. +49
    -50
      consume/init.go
  4. +10
    -0
      consume/md/consume_key.go
  5. +29
    -0
      consume/md/md_user.go
  6. +1
    -1
      go.mod

+ 1
- 1
app/db/dbs_map.go View File

@@ -111,7 +111,7 @@ func GetAllDatabaseDev() *[]model.DbMapping {
if cfg.Local { // 本地调试 加快速度
fmt.Println("notice:LOCAL TEST, only masterId:** 123456 ** available!")
err = Db.Where("deleted_at != ? AND is_dev = '1' AND db_master_id= ?", 1, 123456).
Or("db_master_id = ?", 25616402).Find(&m)
Or("db_master_id = ?", 71353282).Find(&m)
} else {
err = Db.Where("deleted_at != ? AND is_dev = '1' ", 1).Find(&m)
}


+ 101
- 0
consume/canal_service_award_dividend_relation_consume.go View File

@@ -0,0 +1,101 @@
package consume

import (
"applet/app/db"
"applet/app/utils/logx"
"applet/consume/md"
zhios_order_relate_utils "code.fnuoos.com/go_rely_warehouse/zyos_go_es.git/utils"
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
"code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/service_award_dividend"
md2 "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/service_award_dividend/md"
"code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models"
"encoding/json"
"errors"
"fmt"
"github.com/streadway/amqp"
"strings"
)

func CanalServiceAwardDividendRelationConsume(queue md.MqQueue) {
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>")
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(">>>>>>>>>>>>>>>>>>CanalServiceAwardDividendRelationConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
handleCanalServiceAwardDividendRelationConsume(res.Body)
//_ = res.Reject
//_ = res.Ack(true)
} else {
panic(errors.New("error getting message"))
}
}
fmt.Println("get msg done")
}

func handleCanalServiceAwardDividendRelationConsume(msg []byte) error {
//1、解析canal采集至mq中queue的数据结构体
var canalMsg *md.CanalServiceAwardDividendRelationMessage[md.CanalServiceAwardDividendRelation]
err := json.Unmarshal(msg, &canalMsg)
if err != nil {
fmt.Println("err::::", err.Error())
return err
}
if canalMsg.Type != md.CanalMsgUpdateSqlType {
return nil
}
masterId := strings.Split(canalMsg.Database, "_")[1]
if masterId != "71353282" {
return nil
}

if len(canalMsg.Old) > 0 && canalMsg.Old[0].VipLevel != "" && canalMsg.Old[0].VipLevel != canalMsg.Data[0].VipLevel {
eg := db.DBs[masterId]
//TODO::删除旧的记录
var parentServiceAwardDividendRelation models.ServiceAwardDividendRelation
_, err1 := eg.Where("uid = ?", canalMsg.Data[0].BindUid).Get(&parentServiceAwardDividendRelation)
if err1 != nil {
return err1
}
parentServiceAwardDividendRelation.AllocationNums--
_, err1 = eg.Where("id=?", parentServiceAwardDividendRelation.Id).Cols("allocation_nums").Update(&parentServiceAwardDividendRelation)
if err1 != nil {
return err1
}

_, err = eg.Where("id = ?", canalMsg.Data[0].Id).Delete(models.ServiceAwardDividendRelation{})
if err != nil {
fmt.Println("err::::", err.Error())
return err
}

err = service_award_dividend.JoinServiceAwardDividendRelation(eg, md2.JoinServiceAwardDividendRelationReq{
MasterId: masterId,
Uid: zhios_order_relate_utils.StrToInt(canalMsg.Data[0].Uid),
ParentUid: zhios_order_relate_utils.StrToInt(canalMsg.Data[0].ParentUid),
RegionId: zhios_order_relate_utils.StrToInt(canalMsg.Data[0].RegionId),
CreateAt: canalMsg.Data[0].CreateAt,
AllocationNums: zhios_order_relate_utils.StrToInt(canalMsg.Data[0].AllocationNums),
})
if err != nil {
fmt.Println("err::::", err.Error())
return err
}
}

return nil
}

+ 49
- 50
consume/init.go View File

@@ -19,50 +19,48 @@ func Init() {
func initConsumes() {
//jobs[consumeMd.ZhiosGuideStoreOrderFunName] = ZhiosGuideStoreOrder

jobs[consumeMd.ZhiosIntegralProxyRechargeFunName] = ZhiosIntegralProxyRecharge
jobs[consumeMd.ZhiosUserUpLvFunName] = ZhiosUserUpLv
jobs[consumeMd.CanalGuideOrderByUserUpLvConsume] = CanalGuideOrderByUserUpLvConsume
jobs[consumeMd.ZhiosOrderFreeFunName] = ZhiosOrderFree
jobs[consumeMd.ZhiosOrderTotalFunName] = ZhiosOrderTotal
jobs[consumeMd.ZhiosOrderTotalSecondFunName] = ZhiosOrderTotalSecond

jobs[consumeMd.ZhiosOrderSettleTotalFunName] = ZhiosSettleTotal
jobs[consumeMd.ZhiosOrderHjyFunName] = ZhiosOrderHjy
jobs[consumeMd.ZhiosOrderBuckleFunName] = ZhiosOrderBuckle

jobs[consumeMd.ZhiosSupplierAfterOrderFunName] = ZhiosSupplierAfterOrder

jobs[consumeMd.ZhiosAppreciationFunName] = ZhiosAppreciation
jobs[consumeMd.ZhiosValidUserFunName] = ZhiosValidUser

jobs[consumeMd.ZhiosAcquisitionConditionFunName] = ZhiosAcquisitionCondition

jobs[consumeMd.DouShenUserRegisterConsumeForOfficialFunName] = DouShenUserRegisterConsumeForOfficial
jobs[consumeMd.DouShenUserRegisterConsumeForOperationCenterFunName] = DouShenUserRegisterConsumeForOperationCenter
jobs[consumeMd.DouShenUserRegisterConsumeForMyRecommenderFunName] = DouShenUserRegisterConsumeForMyRecommender
jobs[consumeMd.DouShenUserRegisterConsumeForMyFansFunName] = DouShenUserRegisterConsumeForMyFans
jobs[consumeMd.DouShenUserRegisterConsumeForUserRegisterUpLvFunName] = DouShenUserRegisterConsumeForUserRegisterUpLv

jobs[consumeMd.ZhiosFastReturnOrderPayFunName] = ZhiosFastReturnOrderPay
jobs[consumeMd.ZhiosFastReturnOrderSuccessFunName] = ZhiosFastReturnOrderSuccess
jobs[consumeMd.ZhiosFastReturnOrderRefundFunName] = ZhiosFastReturnOrderRefund
jobs[consumeMd.ZhiosFastReturnOrderRefundSecondFunName] = ZhiosFastReturnOrderRefundSecond

jobs[consumeMd.YoumishangExchangeStoreFunName] = YoumishangExchangeStore

jobs[consumeMd.ZhiosRechargeOrderFailFunName] = ZhiosRechargeOrderFail

jobs[consumeMd.CloudIssuanceAsyncMLoginFunName] = CloudIssuanceAsyncMLoginConsume
jobs[consumeMd.ZhiosTikTokUpdateFunName] = ZhiosTikTokUpdate
jobs[consumeMd.ZhiosTikTokAllUpdateFunName] = ZhiosTikTokAllUpdate

jobs[consumeMd.ZhiosCapitalPoolOrderTotalFunName] = ZhiosCapitalPoolOrderTotal
jobs[consumeMd.ZhiosExpressOrderFail] = ZhiosExpressOrderFail
jobs[consumeMd.ZhiosWithdrawReward] = ZhiosWithdrawReward
jobs[consumeMd.ZhiosTaskTotal] = ZhiosTaskTotal
jobs[consumeMd.ZhiosAutoUnFreeze] = ZhiosAutoUnFreeze

//jobs[consumeMd.ZhiosIntegralProxyRechargeFunName] = ZhiosIntegralProxyRecharge
//jobs[consumeMd.ZhiosUserUpLvFunName] = ZhiosUserUpLv
//jobs[consumeMd.CanalGuideOrderByUserUpLvConsume] = CanalGuideOrderByUserUpLvConsume
//jobs[consumeMd.ZhiosOrderFreeFunName] = ZhiosOrderFree
//jobs[consumeMd.ZhiosOrderTotalFunName] = ZhiosOrderTotal
//jobs[consumeMd.ZhiosOrderTotalSecondFunName] = ZhiosOrderTotalSecond
//
//jobs[consumeMd.ZhiosOrderSettleTotalFunName] = ZhiosSettleTotal
//jobs[consumeMd.ZhiosOrderHjyFunName] = ZhiosOrderHjy
//jobs[consumeMd.ZhiosOrderBuckleFunName] = ZhiosOrderBuckle
//
//jobs[consumeMd.ZhiosSupplierAfterOrderFunName] = ZhiosSupplierAfterOrder
//
//jobs[consumeMd.ZhiosAppreciationFunName] = ZhiosAppreciation
//jobs[consumeMd.ZhiosValidUserFunName] = ZhiosValidUser
//
//jobs[consumeMd.ZhiosAcquisitionConditionFunName] = ZhiosAcquisitionCondition
//
//jobs[consumeMd.DouShenUserRegisterConsumeForOfficialFunName] = DouShenUserRegisterConsumeForOfficial
//jobs[consumeMd.DouShenUserRegisterConsumeForOperationCenterFunName] = DouShenUserRegisterConsumeForOperationCenter
//jobs[consumeMd.DouShenUserRegisterConsumeForMyRecommenderFunName] = DouShenUserRegisterConsumeForMyRecommender
//jobs[consumeMd.DouShenUserRegisterConsumeForMyFansFunName] = DouShenUserRegisterConsumeForMyFans
//jobs[consumeMd.DouShenUserRegisterConsumeForUserRegisterUpLvFunName] = DouShenUserRegisterConsumeForUserRegisterUpLv
//
//jobs[consumeMd.ZhiosFastReturnOrderPayFunName] = ZhiosFastReturnOrderPay
//jobs[consumeMd.ZhiosFastReturnOrderSuccessFunName] = ZhiosFastReturnOrderSuccess
//jobs[consumeMd.ZhiosFastReturnOrderRefundFunName] = ZhiosFastReturnOrderRefund
//jobs[consumeMd.ZhiosFastReturnOrderRefundSecondFunName] = ZhiosFastReturnOrderRefundSecond
//
//jobs[consumeMd.YoumishangExchangeStoreFunName] = YoumishangExchangeStore
//
//jobs[consumeMd.ZhiosRechargeOrderFailFunName] = ZhiosRechargeOrderFail
//
//jobs[consumeMd.CloudIssuanceAsyncMLoginFunName] = CloudIssuanceAsyncMLoginConsume
//jobs[consumeMd.ZhiosTikTokUpdateFunName] = ZhiosTikTokUpdate
//jobs[consumeMd.ZhiosTikTokAllUpdateFunName] = ZhiosTikTokAllUpdate
//
//jobs[consumeMd.ZhiosCapitalPoolOrderTotalFunName] = ZhiosCapitalPoolOrderTotal
//jobs[consumeMd.ZhiosExpressOrderFail] = ZhiosExpressOrderFail
//jobs[consumeMd.ZhiosWithdrawReward] = ZhiosWithdrawReward
//jobs[consumeMd.ZhiosTaskTotal] = ZhiosTaskTotal
//jobs[consumeMd.ZhiosAutoUnFreeze] = ZhiosAutoUnFreeze

////////////////////////////////////// V1 /////////////////////////////////////////////////////
//jobs[consumeMd.CloudIssuanceMsgCallBackFunName] = CloudIssuanceMsgCallBackConsume
@@ -96,13 +94,14 @@ func initConsumes() {
//jobs[consumeMd.OneCirclesDealUserCoinConsumeFunName] = OneCirclesDealUserCoinConsume

//////////////////////////////////////// withdraw /////////////////////////////////////////////////////
//jobs[consumeMd.WithdrawConsumeFunName] = WithdrawConsume
//jobs[consumeMd.FlexibleEmploymentWithdrawForGongMaoConsumeFunName] = FlexibleEmploymentWithdrawForGongMaoConsume
//jobs[consumeMd.FlexibleEmploymentWithdrawForPupiaoConsumeFunName] = FlexibleEmploymentWithdrawForPupiaoConsume
//jobs[consumeMd.NewFlexibleEmploymentWithdrawForPupiaoConsumeFunName] = NewFlexibleEmploymentWithdrawForPupiaoConsume
//jobs[consumeMd.ServiceAwardDividendRelationConsumeFunName] = ServiceAwardDividendRelationConsume
//jobs[consumeMd.DividendRewardConsumeFunName] = DividendRewardConsume //蜂蜜分红
//jobs[consumeMd.ContributionExchangeConsumeFunName] = ContributionExchangeConsume //蜂蜜分红-兑换贡献值
jobs[consumeMd.WithdrawConsumeFunName] = WithdrawConsume
jobs[consumeMd.FlexibleEmploymentWithdrawForGongMaoConsumeFunName] = FlexibleEmploymentWithdrawForGongMaoConsume
jobs[consumeMd.FlexibleEmploymentWithdrawForPupiaoConsumeFunName] = FlexibleEmploymentWithdrawForPupiaoConsume
jobs[consumeMd.NewFlexibleEmploymentWithdrawForPupiaoConsumeFunName] = NewFlexibleEmploymentWithdrawForPupiaoConsume
jobs[consumeMd.ServiceAwardDividendRelationConsumeFunName] = ServiceAwardDividendRelationConsume
jobs[consumeMd.CanalServiceAwardDividendRelationConsumeFunName] = CanalServiceAwardDividendRelationConsume //服务奖消费
jobs[consumeMd.DividendRewardConsumeFunName] = DividendRewardConsume //蜂蜜分红
jobs[consumeMd.ContributionExchangeConsumeFunName] = ContributionExchangeConsume //蜂蜜分红-兑换贡献值

//jobs[consumeMd.ZhiosMallGreenCoinConsumeFunName] = ZhiosMallGreenCoinConsume //绿色双链积分
//jobs[consumeMd.ZhiosOneCirclesCoinConsumeFunName] = ZhiosOneCirclesCoinConsume //一个圈圈虚拟币变化


+ 10
- 0
consume/md/consume_key.go View File

@@ -506,6 +506,15 @@ var RabbitMqQueueKeyList = []*MqQueue{
BindKey: "",
ConsumeFunName: "CanalUserVirtualCoinFlowConsume",
},
{
ExchangeName: "canal.topic",
Name: "canal_service_award_dividend_relation",
Type: TopicQueueType,
IsPersistent: false,
RoutKey: "canal_service_award_dividend_relation",
BindKey: "",
ConsumeFunName: "CanalServiceAwardDividendRelationConsume",
},
{
ExchangeName: "one.circles",
Name: "one_circles_sign_in_green_energy",
@@ -713,6 +722,7 @@ const (
InstallmentPaymentAutoRepaidConsumeFunName = "InstallmentPaymentAutoRepaidConsume"
CanalGimMessageConsumeFunName = "CanalGimMessageConsume"
ServiceAwardDividendRelationConsumeFunName = "ServiceAwardDividendRelationConsume"
CanalServiceAwardDividendRelationConsumeFunName = "CanalServiceAwardDividendRelationConsume"
DividendRewardConsumeFunName = "DividendRewardConsume"
ContributionExchangeConsumeFunName = "ContributionExchangeConsume"
)

+ 29
- 0
consume/md/md_user.go View File

@@ -0,0 +1,29 @@
package md

type CanalServiceAwardDividendRelation struct {
Id string `json:"id"`
Uid string `json:"uid"`
ParentUid string `json:"parent_uid"`
BindUid string `json:"bind_uid"`
VipLevel string `json:"vip_level"`
AllocationNums string `json:"allocation_nums"`
RegionId string `json:"region_id"`
CreateAt string `json:"create_at"`
UpdateAt string `json:"update_at"`
}

type CanalServiceAwardDividendRelationMessage[T any] struct {
Data []T `json:"data"`
Database string `json:"database"`
ES int64 `json:"es"`
ID int64 `json:"id"`
IsDdl bool `json:"isDdl"`
Old []struct {
VipLevel string `json:"vip_level"`
UpdateAt string `json:"update_at"`
} `json:"old"`
PkNames []string `json:"pkNames"`
Table string `json:"table"`
TS int64 `json:"ts"`
Type string `json:"type"`
}

+ 1
- 1
go.mod View File

@@ -9,7 +9,7 @@ require (
code.fnuoos.com/go_rely_warehouse/zyos_go_condition_statistics.git v1.1.2-0.20240911091427-ab0c3e964b56
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.5
code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240903060255-62e7a9ea46fe
code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240920031322-e7281ba066ab
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.20240730013105-a13176f971e9
code.fnuoos.com/go_rely_warehouse/zyos_model.git v0.0.4-0.20240830093306-94cbf0dfb7bd


Write Preview
Loading…
Cancel
Save