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

update

one_orenge
DengBiao 10 months ago
parent
08c8cf6ad3
commit
6c375db281
3 changed files with 46 additions and 113 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -68
      consume/canal_order_consume.go
  2. +43
    -43
      consume/init.go
  3. +2
    -2
      consume/zhios_user_visit_ip_address_consume.go

+ 1
- 68
consume/canal_order_consume.go View File

@@ -92,73 +92,6 @@ func CanalOrderConsumev2(queue md.MqQueue) {
}

func CanalOrderConsume(queue md.MqQueue) {
//now := time.Now()
//var list []model2.Group
//err := db.ImDb.Where("master_id = 68823769").And("user_num > ?", 500).Find(&list)
////eg := db.DBs["68823769"]
//for _, v := range list {
// var groupUsers []model2.GroupUser
// var groupLeader model2.GroupUser
// err = db.ImDb.Where("group_id = ?", v.Id).And("member_type = 2").Limit(10000, 499).Find(&groupUsers)
// db.ImDb.Where("group_id = ?", v.Id).And("member_type = 1").Get(&groupLeader)
//
// //新增一个群
// re := regexp.MustCompile("[0-9]+")
// num := re.FindAllString(v.Name, -1)[0]
// name := strings.Replace(v.Name, num, strconv.Itoa(utils.StrToInt(num)+1), 1)
// temGroup := model2.Group{
// Name: name,
// AvatarUrl: "",
// Introduction: "",
// UserNum: int32(len(groupUsers)) + 1,
// IsAllMemberBanned: 0,
// MasterId: v.MasterId,
// Extra: "",
// CreateTime: now,
// UpdateTime: now,
// }
// _, err := db.ImDb.InsertOne(&temGroup)
// if err != nil {
// panic(err)
// }
// groupId := temGroup.Id
//
// //新增群主
// _, err = db.ImDb.InsertOne(model2.GroupUser{
// GroupId: groupId,
// UserId: groupLeader.UserId,
// MemberType: 1,
// Status: 0,
// CreateTime: now.Format("2006-01-02 15:04:05"),
// UpdateTime: now.Format("2006-01-02 15:04:05"),
// })
// if err != nil {
// panic(err)
// }
//
// for _, groupUser := range groupUsers {
// groupUser.GroupId = groupId
// db.ImDb.ID(groupUser.Id).Update(groupUser)
// }
// v.UserNum = 500
// db.ImDb.ID(v.Id).Update(v)
// var oldDouShenImGroup model.DouShenImGroup
// if has, err := db.Db.Where("group_id =?", v.Id).Get(&oldDouShenImGroup); err != nil || has == false {
// panic(err)
// }
// _, err = db.DouShenImGroupInsert(db.Db, &model.DouShenImGroup{
// Kind: oldDouShenImGroup.Kind,
// Uid: int(oldDouShenImGroup.Uid),
// GroupId: int(groupId),
// IsFull: 0,
// Name: name,
// CreateTime: now,
// UpdateTime: now,
// })
//}
//
//return
//fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>")
ch, err := rabbit.Cfg.Pool.GetChannel()
if err != nil {
logx.Error(err)
@@ -168,7 +101,7 @@ func CanalOrderConsume(queue md.MqQueue) {
//1、将自己绑定到交换机上
ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
//2、取出数据进行消费
ch.Qos(1)
ch.Qos(100)
delivery := ch.Consume(queue.Name, false)

var res amqp.Delivery


+ 43
- 43
consume/init.go View File

@@ -17,47 +17,47 @@ func Init() {

// 增加消费任务队列
func initConsumes() {
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.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.ZhiosGuideStoreOrderFunName] = ZhiosGuideStoreOrder
//
jobs[consumeMd.ZhiosOrderSettleTotalFunName] = ZhiosSettleTotal
jobs[consumeMd.ZhiosOrderHjyFunName] = ZhiosOrderHjy
jobs[consumeMd.ZhiosOrderBuckleFunName] = ZhiosOrderBuckle
//jobs[consumeMd.ZhiosAppreciationFunName] = ZhiosAppreciation
//jobs[consumeMd.ZhiosValidUserFunName] = ZhiosValidUser
//
jobs[consumeMd.ZhiosSupplierAfterOrderFunName] = ZhiosSupplierAfterOrder
jobs[consumeMd.ZhiosGuideStoreOrderFunName] = ZhiosGuideStoreOrder

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.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
//

////////////////////////////////////// V1 /////////////////////////////////////////////////////
@@ -70,10 +70,10 @@ 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
}

func Run() {


+ 2
- 2
consume/zhios_user_visit_ip_address_consume.go View File

@@ -22,7 +22,7 @@ import (
var data []*model.MasterAreaVisitsFlows

func ZhiOsUserVisitIpAddressConsume(queue md.MqQueue) {
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>")
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>ZhiOsUserVisitIpAddressConsume<<<<<<<<<<<<<<<<<<<<<<<")
ch, err := rabbit.Cfg.Pool.GetChannel()
if err != nil {
logx.Error(err)
@@ -35,7 +35,7 @@ func ZhiOsUserVisitIpAddressConsume(queue md.MqQueue) {
//1、将自己绑定到交换机上
ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
//2、取出数据进行消费
ch.Qos(10)
ch.Qos(100)
delivery := ch.Consume(queue.Name, false)

var res amqp.Delivery


Write Preview
Loading…
Cancel
Save