common/constant.go

@@ -319,7 +319,6 @@ const (
 	GainWayItemBagChangeDoll                 = 111 // 背包内兑换娃娃
 	GainWayClawdollCatch                     = 112 // 娃娃机抓取到娃娃获取卡
 	GainWayItemBagChangeDollRevocation       = 113 //娃娃兑换后台撤销
-	GainWayPermitReset                       = 114 //赛季通行证积分重置
 )
 
 // 后台选择 金币变化类型 的充值 类型id号起始

dbproxy/main.go

@@ -12,7 +12,7 @@ import (
 
 	_ "mongo.games.com/game"
 	"mongo.games.com/game/common"
-	_ "mongo.games.com/game/dbproxy/mq"
+	dbmq "mongo.games.com/game/dbproxy/mq"
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
@@ -20,6 +20,7 @@ import (
 )
 
 func main() {
+	dbmq.Init()
 	// 自定义配置文件
 	model.InitGameParam()
 	// package模块

dbproxy/mq/c_apilog.go

@@ -1,25 +1,36 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.DBAPILog,
-		Data: model.APILog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.APILog)
-			if !ok {
-				return nil
+	mq.RegisterSubscriber(mq.MgrAPILog, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.APILog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
+				return
 			}
+
 			c := svc.APILogsCollection()
 			if c != nil {
 				err = c.Insert(log)
 			}
-			return nil
-		},
-	})
+			return
+		}
+		return nil
+	}, broker.Queue(mq.MgrAPILog), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_coingivelog.go

@@ -1,25 +1,36 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.DBGiveLog,
-		Data: model.CoinGiveLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.CoinGiveLog)
-			if !ok {
+	mq.RegisterSubscriber(model.CoinGiveLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.CoinGiveLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.CoinGiveLogCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.CoinGiveLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_coinlog.go

@@ -1,20 +1,30 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.CoinLogCollName,
-		Data: &model.CoinLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.CoinLog)
-			if !ok {
+	mq.RegisterSubscriber(model.CoinLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.CoinLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			if log.Count == 0 { //玩家冲账探针
 				mq.Send(mq.WorldLogCoinAck, log)
 			} else {
@@ -30,6 +40,7 @@ func init() {
 				}
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.CoinLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_customlog.go

@@ -7,12 +7,6 @@ import (
 )
 
 func init() {
-	// 竞技馆获奖通知
-	mq.RegisterMessage(&mq.RegisterMessageParam{
-		Name: mq.RankCustomLogAward,
-		Data: model.CustomLogAward{},
-	})
-
 	// 竞技馆对局记录
 	mq.RegisterHandler(&mq.RegisterHandlerParam{
 		Name: mq.DBCustomLog,

dbproxy/mq/c_friendrecordlog.go

@@ -1,22 +1,32 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.FriendRecordLogCollName,
-		Data: &model.FriendRecord{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.FriendRecord)
-			if !ok {
+	mq.RegisterSubscriber(model.FriendRecordLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.FriendRecord
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
-			err = svc.InsertFriendRecordLog(*log)
+			err = svc.InsertFriendRecordLog(log)
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.FriendRecordLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_gamegamedetailedlog.go

@@ -1,6 +1,10 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
 
 	"mongo.games.com/game/dbproxy/svc"
@@ -9,14 +13,19 @@ import (
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.GameDetailedLogCollName,
-		Data: &model.GameDetailedLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.GameDetailedLog)
-			if !ok {
+	mq.RegisterSubscriber(model.GameDetailedLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.GameDetailedLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+			logger.Logger.Tracef("mq receive GameDetailedLog:%v", log)
 			c := svc.GameDetailedLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
@@ -25,6 +34,7 @@ func init() {
 				}
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.GameDetailedLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_gameplayerlistlog.go

@@ -1,25 +1,36 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.GamePlayerListLogCollName,
-		Data: &model.GamePlayerListLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.GamePlayerListLog)
-			if !ok {
+	mq.RegisterSubscriber(model.GamePlayerListLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.GamePlayerListLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.GamePlayerListLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.GamePlayerListLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_invite.go

@@ -2,11 +2,14 @@ package mq
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
 	newMongo "go.mongodb.org/mongo-driver/mongo"
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
 
 	"mongo.games.com/game/common"
@@ -17,14 +20,21 @@ import (
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.DBInvite,
-		Data: &model.EvtInviteMsg{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.EvtInviteMsg)
-			if !ok {
+	mq.RegisterSubscriber(mq.DBInvite, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.EvtInviteMsg
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
+				logger.Logger.Errorf("EvtInvite json.Unmarshal error:%v msg:%v", err, string(msg.Body))
 				return
 			}
+			logger.Logger.Tracef("EvtInvite log:%+v", log)
+
 			var addRechargeScore bool
 			if log.Tp != common.InviteScoreCheckWeek {
 				addRechargeScore, err = svc.CheckInviteScore(&log.InviteScore)
@@ -293,7 +303,8 @@ func init() {
 				}
 				mq.Send(mq.WorldInviteAck, msg)
 			}
-			return
-		},
-	})
+
+		}
+		return nil
+	}, broker.Queue(mq.DBInvite), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_itemlog.go

@@ -1,25 +1,36 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+	
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.ItemLogCollName,
-		Data: &model.ItemLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.ItemLog)
-			if !ok {
+	mq.RegisterSubscriber(model.ItemLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.ItemLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.ItemLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.ItemLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_loginlog.go

@@ -1,27 +1,38 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.LoginLogCollName,
-		Data: &model.LoginLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.LoginLog)
-			if !ok {
+	mq.RegisterSubscriber(model.LoginLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.LoginLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.LoginLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.LoginLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 
 	//for test
 	//RegisterSubscriber(model.LoginLogCollName, func(e broker.Event) (err error) {

dbproxy/mq/c_onlinelog.go

@@ -1,26 +1,37 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.OnlineLogCollName,
-		Data: &model.OnlineLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.OnlineLog)
-			if !ok {
+	mq.RegisterSubscriber(model.OnlineLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.OnlineLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := new(svc.OnlineLogSvc)
 			if c != nil {
 				ret := false
-				err = c.InsertSignleOnlineLog(log, &ret)
+				err = c.InsertSignleOnlineLog(&log, &ret)
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.OnlineLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_rank.go

@@ -1,6 +1,12 @@
 package mq
 
 import (
+	"encoding/json"
+
+	"mongo.games.com/goserver/core"
+	"mongo.games.com/goserver/core/basic"
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
 
 	"mongo.games.com/game/dbproxy/svc"
@@ -10,70 +16,117 @@ import (
 
 func init() {
 	// 排位排行榜
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.MQRankSeason,
-		Data: &model.PlayerRankScore{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.PlayerRankScore)
-			if !ok {
+	mq.RegisterSubscriber(model.MQRankSeason, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.PlayerRankScore
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
-			err = svc.RankSeasonUpsert(log)
-			if err != nil {
-				logger.Logger.Errorf("RankSeasonUpsert err: %v", err)
-			}
+
+			logger.Logger.Tracef("SubscriberPlayerRankScore: %+v", log)
+
+			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
+				err := svc.RankSeasonUpsert(&log)
+				if err != nil {
+					logger.Logger.Errorf("RankSeasonUpsert err: %v", err)
+				}
+				return nil
+			}), true)
+
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.MQRankSeason), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 
 	// 金币榜
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.MQRankPlayerCoin,
-		Data: &model.RankPlayerCoin{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.RankPlayerCoin)
-			if !ok {
+	mq.RegisterSubscriber(model.MQRankPlayerCoin, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.RankPlayerCoin
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
-			err = svc.RankPlayerCoinUpsert(log)
-			if err != nil {
-				logger.Logger.Errorf("RankPlayerCoinUpsert err: %v", err)
-			}
+
+			logger.Logger.Tracef("SubscriberRankPlayerCoin: %+v", log)
+
+			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
+				err := svc.RankPlayerCoinUpsert(&log)
+				if err != nil {
+					logger.Logger.Errorf("RankPlayerCoinUpsert err: %v", err)
+				}
+				return nil
+			}), true)
+
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.MQRankPlayerCoin), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 
 	//等级榜
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.MQRankPlayerLevel,
-		Data: &model.PlayerLevelInfo{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.PlayerLevelInfo)
-			if !ok {
-				return
-			}
-			err = svc.RankPlayerLevelUpsert(log)
-			if err != nil {
-				logger.Logger.Errorf("RankPlayerLevelUpsert err: %v", err)
-			}
-			return
-		},
-	})
+	mq.RegisterSubscriber(model.MQRankPlayerLevel, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
 
-	// 赛季通行证积分
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.MQRankPlayerPermit,
-		Data: &model.PermitScore{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.PermitScore)
-			if !ok {
+			var log model.PlayerLevelInfo
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
-			err = svc.RankPlayerPermitUpsert(log)
-			if err != nil {
-				logger.Logger.Errorf("RankPlayerPermitUpsert err: %v", err)
-			}
+
+			logger.Logger.Tracef("SubscriberRankPlayerLevel: %+v", log)
+
+			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
+				err := svc.RankPlayerLevelUpsert(&log)
+				if err != nil {
+					logger.Logger.Errorf("RankPlayerCoinUpsert err: %v", err)
+				}
+				return nil
+			}), true)
+
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.MQRankPlayerLevel), broker.DisableAutoAck(), rabbitmq.DurableQueue())
+	// 赛季通行证积分
+	mq.RegisterSubscriber(model.MQRankPlayerPermit, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.PermitScore
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
+				return
+			}
+
+			logger.Logger.Tracef("SubscriberRankPlayerPermit: %+v", log)
+
+			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
+				err := svc.RankPlayerPermitUpsert(&log)
+				if err != nil {
+					logger.Logger.Errorf("RankPlayerPermitUpsert err: %v", err)
+				}
+				return nil
+			}), true)
+
+			return
+		}
+		return nil
+	}, broker.Queue(model.MQRankPlayerPermit), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_scenecoinlog.go

@@ -1,7 +1,10 @@
 package mq
 
 import (
-	"mongo.games.com/goserver/core/logger"
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -9,22 +12,25 @@ import (
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.SceneCoinLogCollName,
-		Data: &model.SceneCoinLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.SceneCoinLog)
-			if !ok {
+	mq.RegisterSubscriber(model.SceneCoinLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.SceneCoinLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.SceneCoinLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
-				if err != nil {
-					logger.Logger.Errorf("insert SceneCoinLog failed, err: %v", err)
-				}
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.SceneCoinLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_vipgiftlog.go

@@ -1,7 +1,10 @@
 package mq
 
 import (
-	"mongo.games.com/goserver/core/logger"
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -9,22 +12,25 @@ import (
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.DBVipGiftLog,
-		Data: &model.DbVip{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.DbVip)
-			if !ok {
+	mq.RegisterSubscriber(mq.DBVipGiftLog, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.DbVip
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.DbVipLogCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
-				if err != nil {
-					logger.Logger.Errorf("insert DBVipGiftLog error:%v", err)
-				}
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(mq.DBVipGiftLog), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_welfarelog.go

@@ -1,7 +1,10 @@
 package mq
 
 import (
-	"mongo.games.com/goserver/core/logger"
+	"encoding/json"
+
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -9,22 +12,25 @@ import (
 )
 
 func init() {
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: model.WelfareLogCollName,
-		Data: &model.WelfareLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.WelfareLog)
-			if !ok {
+	mq.RegisterSubscriber(model.WelfareLogCollName, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.WelfareLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			c := svc.WelfareLogsCollection(log.Platform)
 			if c != nil {
 				err = c.Insert(log)
-				if err != nil {
-					logger.Logger.Errorf("insert WelfareLog failed: %v", err)
-				}
 			}
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(model.WelfareLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/init.go

@@ -0,0 +1,19 @@
+package mq
+
+import (
+	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
+)
+
+func Init() {
+	InitHandler()
+	// 竞技馆获奖通知
+	mq.RegisterMessage(&mq.RegisterMessageParam{
+		Name: mq.RankCustomLogAward,
+		Data: model.CustomLogAward{},
+	})
+}
+
+func InitHandler() {
+
+}

gamesrv/base/logchannel.go

@@ -0,0 +1,62 @@
+package base
+
+import (
+	"reflect"
+
+	"mongo.games.com/goserver/core/logger"
+
+	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
+)
+
+// LogChannelSingleton 日志记录器
+var LogChannelSingleton = &LogChannel{
+	cName: make(map[reflect.Type]string),
+}
+
+type LogChannel struct {
+	cName map[reflect.Type]string
+}
+
+func (c *LogChannel) RegisterLogCName(cname string, log interface{}) {
+	t := c.getLogType(log)
+	c.cName[t] = cname
+}
+
+func (c *LogChannel) getLogType(log interface{}) reflect.Type {
+	return reflect.Indirect(reflect.ValueOf(log)).Type()
+}
+
+func (c *LogChannel) getLogCName(log interface{}) string {
+	t := c.getLogType(log)
+	if name, exist := c.cName[t]; exist {
+		return name
+	}
+	return ""
+}
+
+func (c *LogChannel) WriteLog(log interface{}) {
+	cname := c.getLogCName(log)
+	if cname == "" {
+		cname = "_null_"
+	}
+	logger.Logger.Tracef("LogChannel ==> %#v", log)
+	mq.Send(cname, log)
+}
+
+func (c *LogChannel) WriteMQData(data *model.RabbitMQData) {
+	mq.Send(data.MQName, data.Data)
+}
+
+func init() {
+	LogChannelSingleton.RegisterLogCName(model.LoginLogCollName, &model.LoginLog{})
+	LogChannelSingleton.RegisterLogCName(model.CoinGiveLogCollName, &model.CoinGiveLog{})
+	LogChannelSingleton.RegisterLogCName(model.CoinLogCollName, &model.CoinLog{})
+	LogChannelSingleton.RegisterLogCName(model.SceneCoinLogCollName, &model.SceneCoinLog{})
+	LogChannelSingleton.RegisterLogCName(model.GameDetailedLogCollName, &model.GameDetailedLog{})
+	LogChannelSingleton.RegisterLogCName(model.GamePlayerListLogCollName, &model.GamePlayerListLog{})
+	LogChannelSingleton.RegisterLogCName(model.FriendRecordLogCollName, &model.FriendRecord{})
+	LogChannelSingleton.RegisterLogCName(model.ItemLogCollName, &model.ItemLog{})
+	LogChannelSingleton.RegisterLogCName(mq.DBCustomLog, &model.CustomLog{})
+	LogChannelSingleton.RegisterLogCName(mq.DBCustomLogAward, &model.CustomLogAward{})
+}

gamesrv/base/player.go

@@ -2,22 +2,21 @@ package base
 
 import (
 	"fmt"
-	"math"
-	"math/rand"
-	"time"
-
 	rawproto "google.golang.org/protobuf/proto"
-	"mongo.games.com/goserver/core/logger"
-	"mongo.games.com/goserver/core/netlib"
-	"mongo.games.com/goserver/core/timer"
-
+	"math"
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/player"
 	"mongo.games.com/game/protocol/server"
 	"mongo.games.com/game/srvdata"
+	"mongo.games.com/goserver/core/logger"
+	"mongo.games.com/goserver/core/netlib"
+	"mongo.games.com/goserver/core/timer"
+
+	//rawproto "github.com/golang/protobuf/proto"
+	"math/rand"
+	"time"
 )
 
 // 对应到客户端的一个玩家对象.
@@ -475,7 +474,7 @@ func (this *Player) AddCoin(num int64, gainWay int32, syncFlag int, oper, remark
 			if log != nil {
 				this.GameCoinTs = log.Time.UnixNano()
 				this.dirty = true
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 			}
 		}
 		//确保金币场金币数量不小于0
@@ -558,7 +557,7 @@ func (this *Player) AddCoinAsync(num int64, gainWay int32, notifyC, broadcast bo
 			if log != nil {
 				this.GameCoinTs = log.Time.UnixNano()
 				this.dirty = true
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 			}
 		}
 		//确保金币场金币数量不小于0
@@ -639,11 +638,11 @@ func (this *Player) ReportGameEvent(tax, taxex, changeCoin, validbet, validFlow,
 	}
 
 	gamingTime := int32(time.Now().Sub(this.scene.GameNowTime).Seconds())
-	mq.Write(model.CreatePlayerGameRecEvent(this.SnId, tax, taxex, changeCoin, validbet, validFlow, in, out,
+	LogChannelSingleton.WriteMQData(model.GenerateGameEvent(model.CreatePlayerGameRecEvent(this.SnId, tax, taxex, changeCoin, validbet, validFlow, in, out,
 		int32(this.scene.GameId), this.scene.GetGameFreeId(), int32(this.scene.GameMode),
 		this.scene.GetRecordId(), this.Channel, this.BeUnderAgentCode, this.Platform, this.City, this.DeviceOS,
 		this.CreateTime, gamingTime, gameFirstTime, gameFreeFirstTime, gameTimes, gameFreeTimes, this.LastLoginTime,
-		this.TelephonePromoter, this.DeviceId), mq.BackGameRecord)
+		this.TelephonePromoter, this.DeviceId)))
 }
 
 // 汇总玩家该次游戏总产生的税收

gamesrv/base/replay_recorder.go

@@ -6,13 +6,11 @@ import (
 	"encoding/json"
 	"time"
 
-	"mongo.games.com/goserver/core/logger"
-	"mongo.games.com/goserver/core/netlib"
-
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/server"
+	"mongo.games.com/goserver/core/logger"
+	"mongo.games.com/goserver/core/netlib"
 )
 
 var _replayIgnorePacketIds = map[int]bool{}
@@ -114,7 +112,7 @@ func (this *ReplayRecorder) Fini(s *Scene) {
 				break
 			}
 		}
-		mq.Write(pack, "log_playback")
+		LogChannelSingleton.WriteMQData(&model.RabbitMQData{MQName: "log_playback", Data: pack})
 	}
 }
 

gamesrv/base/scene.go

@@ -16,7 +16,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/gamehall"
 	"mongo.games.com/game/protocol/player"
@@ -1436,7 +1435,7 @@ func (this *Scene) SaveGameDetailedLog(logid string, gamedetailednote string, ga
 							if this.IsCustom() {
 								log.CycleId = this.CycleID
 							}
-							mq.Write(log)
+							LogChannelSingleton.WriteLog(log)
 						}
 					}
 				}
@@ -1455,7 +1454,7 @@ func (this *Scene) SaveGameDetailedLog(logid string, gamedetailednote string, ga
 					}
 					newLog := new(model.GameDetailedLog)
 					*newLog = *log
-					mq.Write(log)
+					LogChannelSingleton.WriteLog(log)
 				}
 			}
 		}
@@ -1515,7 +1514,7 @@ func (this *Scene) SaveFriendRecord(snid int32, isWin int32, billCoin int64, bas
 	}
 	log := model.NewFriendRecordLogEx(this.Platform, snid, isWin, this.GameId, baseScore, billCoin, int64(this.GetMatch().GetMatchType()))
 	if log != nil {
-		mq.Write(log)
+		LogChannelSingleton.WriteLog(log)
 	}
 }
 
@@ -1564,7 +1563,7 @@ func (this *Scene) SaveGamePlayerListLog(snid int32, param *SaveGamePlayerListLo
 				param.BetAmount, param.WinAmountNoAnyTax, this.KeyGameId, Name, this.GetDBGameFree().GetGameClass(),
 				param.IsFirstGame, this.GetMatch().GetMatchSortId(), int64(this.GetMatch().GetMatchType()), param.IsFree, param.WinSmallGame, param.WinTotal, param.CycleId)
 			if log != nil {
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 			}
 		}
 	}

gamesrv/main.go

@@ -12,7 +12,6 @@ import (
 	"mongo.games.com/game/common"
 	_ "mongo.games.com/game/gamesrv/action"
 	_ "mongo.games.com/game/gamesrv/base"
-	_ "mongo.games.com/game/gamesrv/mq"
 	_ "mongo.games.com/game/gamesrv/transact"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"

gamesrv/mq/init.go

@@ -1,19 +0,0 @@
-package mq
-
-import (
-	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
-)
-
-func init() {
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.LoginLogCollName, Data: &model.LoginLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBGiveLog, Data: &model.CoinGiveLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.CoinLogCollName, Data: &model.CoinLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.SceneCoinLogCollName, Data: &model.SceneCoinLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.GameDetailedLogCollName, Data: &model.GameDetailedLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.GamePlayerListLogCollName, Data: &model.GamePlayerListLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.FriendRecordLogCollName, Data: &model.FriendRecord{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.ItemLogCollName, Data: &model.ItemLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBCustomLog, Data: &model.CustomLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBCustomLogAward, Data: &model.CustomLogAward{}})
-}

gamesrv/tienlen/scenedata_tienlen.go

@@ -16,7 +16,6 @@ import (
 	rule "mongo.games.com/game/gamerule/tienlen"
 	"mongo.games.com/game/gamesrv/base"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/tienlen"
 	"mongo.games.com/game/srvdata"
@@ -2164,5 +2163,5 @@ func (this *TienLenSceneData) SaveCustomLog() {
 			LogId: this.RoundLogId[k],
 		})
 	}
-	mq.Write(log)
+	base.LogChannelSingleton.WriteLog(log)
 }

gamesrv/tienlen/scenepolicy_tienlen.go

@@ -14,7 +14,6 @@ import (
 	rule "mongo.games.com/game/gamerule/tienlen"
 	"mongo.games.com/game/gamesrv/base"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/tienlen"
 	"mongo.games.com/game/srvdata"
@@ -2640,7 +2639,7 @@ func (this *SceneBilledStateTienLen) OnEnter(s *base.Scene) {
 								GameId:     int64(sceneEx.GameId),
 								GameFreeId: int64(sceneEx.GetGameFreeId()),
 							})
-							mq.Write(&model.CustomLogAward{
+							base.LogChannelSingleton.WriteLog(&model.CustomLogAward{
 								Platform: p.Platform,
 								CycleId:  sceneEx.CycleID,
 								SnId:     p.SnId,

gatesrv/clientlog.go

@@ -13,6 +13,8 @@ import (
 )
 
 func init() {
+	LogChannelSingleton.RegisterLogCName(mq.BackClientLog, &model.ClientLogMysql{})
+
 	// 客户端日志
 	netlib.Register(int(player.PlayerPacketID_PACKET_CS_CLIENT_LOG), player.CSClientLog{}, CSClientLog)
 }
@@ -53,7 +55,7 @@ func CSClientLog(s *netlib.Session, packetid int, pack interface{}) error {
 	//	Ts:       time.Now().Unix(),
 	//})
 
-	mq.Write(&model.ClientLogMysql{
+	LogChannelSingleton.WriteLog(&model.ClientLogMysql{
 		Data:       msg.GetData(),
 		Platform:   platform,
 		Snid:       snid,
@@ -62,5 +64,6 @@ func CSClientLog(s *netlib.Session, packetid int, pack interface{}) error {
 		AppChannel: appChannel,
 		ChannelId:  channelId,
 	})
+
 	return nil
 }

gatesrv/logchannel.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"reflect"
+
+	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
+)
+
+// LogChannelSingleton 日志记录器
+var LogChannelSingleton = &LogChannel{
+	cName: make(map[reflect.Type]string),
+}
+
+type LogChannel struct {
+	cName map[reflect.Type]string
+}
+
+func (c *LogChannel) RegisterLogCName(cname string, log interface{}) {
+	t := c.getLogType(log)
+	c.cName[t] = cname
+}
+
+func (c *LogChannel) getLogType(log interface{}) reflect.Type {
+	return reflect.Indirect(reflect.ValueOf(log)).Type()
+}
+
+func (c *LogChannel) getLogCName(log interface{}) string {
+	t := c.getLogType(log)
+	if name, exist := c.cName[t]; exist {
+		return name
+	}
+	return ""
+}
+
+func (c *LogChannel) WriteLog(log interface{}) {
+	cname := c.getLogCName(log)
+	if cname == "" {
+		cname = "_null_"
+	}
+	mq.Send(cname, log)
+}
+
+func (c *LogChannel) WriteMQData(data *model.RabbitMQData) {
+	mq.Send(data.MQName, data.Data)
+}

gatesrv/main.go

@@ -6,7 +6,6 @@ import (
 
 	_ "mongo.games.com/game"
 	"mongo.games.com/game/common"
-	_ "mongo.games.com/game/gatesrv/mq"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 )

gatesrv/mq/init.go

@@ -1,10 +0,0 @@
-package mq
-
-import (
-	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
-)
-
-func init() {
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackClientLog, Data: &model.ClientLogMysql{}})
-}

mgrsrv/api/logchannel.go

@@ -0,0 +1,57 @@
+package api
+
+import (
+	"reflect"
+
+	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
+)
+
+/*
+ api调用记录写入rabbitmq,然后dbproxy收到rabbitmq数据再写到数据库
+*/
+
+const (
+	LogChanelBlackHole = "_null_"
+)
+
+var LogChannelSington = &LogChannel{
+	cName: make(map[reflect.Type]string),
+}
+
+type LogChannel struct {
+	cName map[reflect.Type]string
+}
+
+func (c *LogChannel) RegisteLogCName(cname string, log interface{}) {
+	t := c.getLogType(log)
+	c.cName[t] = cname
+}
+
+func (c *LogChannel) getLogType(log interface{}) reflect.Type {
+	return reflect.Indirect(reflect.ValueOf(log)).Type()
+}
+
+func (c *LogChannel) getLogCName(log interface{}) string {
+	t := c.getLogType(log)
+	if name, exist := c.cName[t]; exist {
+		return name
+	}
+	return ""
+}
+
+func (c *LogChannel) WriteLog(log interface{}) {
+	cname := c.getLogCName(log)
+	if cname == "" {
+		cname = LogChanelBlackHole
+	}
+	mq.Send(cname, log)
+}
+
+func (c *LogChannel) WriteMQData(data *model.RabbitMQData) {
+	mq.Send(data.MQName, data.Data)
+}
+
+func init() {
+	LogChannelSington.RegisteLogCName(mq.MgrAPILog, &model.APILog{})
+}

mgrsrv/api/webapi_gamesrv.go

@@ -20,7 +20,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/webapi"
 )
@@ -122,7 +121,7 @@ func GameSrvWebAPI(rw http.ResponseWriter, req *http.Request) {
 	result, err := json.Marshal(rep)
 	if err == nil {
 		log := model.NewAPILog(req.URL.Path, req.URL.RawQuery, string(data[:]), req.RemoteAddr, string(result[:]), startTime, ps)
-		mq.Write(log)
+		LogChannelSington.WriteLog(log)
 	}
 	return
 }

mgrsrv/api/webapi_worldsrv.go

@@ -21,7 +21,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 )
 
 // API
@@ -134,7 +133,7 @@ func WorldSrvApi(rw http.ResponseWriter, req *http.Request) {
 	result, err := json.Marshal(rep)
 	if err == nil {
 		log := model.NewAPILog(req.URL.Path, req.URL.RawQuery, string(data[:]), req.RemoteAddr, string(result[:]), startTime, ps)
-		mq.Write(log)
+		LogChannelSington.WriteLog(log)
 	}
 	return
 }

mgrsrv/main.go

@@ -9,7 +9,6 @@ import (
 	_ "mongo.games.com/game"
 	"mongo.games.com/game/common"
 	_ "mongo.games.com/game/mgrsrv/api"
-	_ "mongo.games.com/game/mgrsrv/mq"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"
 	_ "mongo.games.com/game/srvdata"

mgrsrv/mq/init.go

@@ -1,10 +0,0 @@
-package mq
-
-import (
-	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
-)
-
-func init() {
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBAPILog, Data: &model.APILog{}})
-}

model/gameparam.go

@@ -89,7 +89,6 @@ type GameParam struct {
 	CustomAwardMaxAddTime      int      // 竞技馆假奖励方法周期，单位秒
 	AdminPassword              string   // 管理员密码
 	UseAdminPassword           bool     // 是否使用管理员密码
-	CloseCustomRoomCreate      bool     // 关闭自定义房间创建
 }
 
 var GameParamPath = "../data/gameparam.json"

model/rabbit_mq.go

@@ -8,7 +8,7 @@ import (
 )
 
 // GenerateOnline 在线统计
-func GenerateOnline(online map[string]map[string]int) *mq.RabbitMQData {
+func GenerateOnline(online map[string]map[string]int) *RabbitMQData {
 	params := make(map[string]interface{})
 	params["Online"] = online
 	params["Time"] = time.Now().Unix()
@@ -16,12 +16,17 @@ func GenerateOnline(online map[string]map[string]int) *mq.RabbitMQData {
 }
 
 // GenerateLogin 玩家登陆事件
-func GenerateLogin(o *PlayerLoginEvent) *mq.RabbitMQData {
+func GenerateLogin(o *PlayerLoginEvent) *RabbitMQData {
 	return NewRabbitMQData(mq.BackLogin, o)
 }
 
+// GenerateGameEvent 游戏结算事件
+func GenerateGameEvent(o *PlayerGameRecEvent) *RabbitMQData {
+	return NewRabbitMQData(mq.BackGameRecord, o)
+}
+
 // GenerateSystemFreeGive 系统免费赠送
-func GenerateSystemFreeGive(snid int32, name, platform, channel string, givetype, cointype int32, count int64) *mq.RabbitMQData {
+func GenerateSystemFreeGive(snid int32, name, platform, channel string, givetype, cointype int32, count int64) *RabbitMQData {
 	params := make(map[string]string)
 	params["snid"] = strconv.Itoa(int(snid))
 	params["Channel"] = channel
@@ -35,7 +40,7 @@ func GenerateSystemFreeGive(snid int32, name, platform, channel string, givetype
 }
 
 // GeneratePhoneLottery 手机抽奖统计
-func GeneratePhoneLottery(snid int32, platform string, items string, lotteryType, count, countType, score int) *mq.RabbitMQData {
+func GeneratePhoneLottery(snid int32, platform string, items string, lotteryType, count, countType, score int) *RabbitMQData {
 	params := make(map[string]string)
 	params["Snid"] = strconv.Itoa(int(snid))
 	params["Platform"] = platform

model/rabbitmqdata.go

@@ -1,9 +1,12 @@
 package model
 
-import "mongo.games.com/game/mq"
+type RabbitMQData struct {
+	MQName string // 是队列名称也是Routing key
+	Data   interface{}
+}
 
-func NewRabbitMQData(mqName string, data interface{}) *mq.RabbitMQData {
-	log := &mq.RabbitMQData{
+func NewRabbitMQData(mqName string, data interface{}) *RabbitMQData {
+	log := &RabbitMQData{
 		MQName: mqName,
 		Data:   data,
 	}

mq/consumer.go

@@ -1,4 +1,4 @@
-package internal
+package mq
 
 import (
 	"fmt"
@@ -10,6 +10,8 @@ import (
 	"mongo.games.com/goserver/core/broker"
 	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
+
+	"mongo.games.com/game/common"
 )
 
 /*
@@ -17,6 +19,7 @@ import (
 */
 
 var (
+	globalConsumer *RabbitMQConsumer
 	subscriberLock sync.RWMutex
 	subscriber     = make(map[string][]*Subscriber)
 )
@@ -53,7 +56,7 @@ func NewRabbitMQConsumer(url string, exchange rabbitmq.Exchange, opts ...broker.
 }
 
 func (c *RabbitMQConsumer) Start() error {
-	if ok, _ := PathExists(BackupPath); !ok {
+	if ok, _ := common.PathExists(BackupPath); !ok {
 		os.MkdirAll(BackupPath, os.ModePerm)
 	}
 
@@ -141,3 +144,19 @@ func GetSubscribers() map[string][]*Subscriber {
 	}
 	return ret
 }
+
+// StartConsumer 启动消费者
+func StartConsumer(url string, exchange string, durableExchange bool, opts ...broker.Option) {
+	StopConsumer()
+	globalConsumer = NewRabbitMQConsumer(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, opts...)
+	if err := globalConsumer.Start(); err != nil {
+		panic(fmt.Sprintf("RabbitMQConsumer.Start() url:%v exchange:%v err:%v", url, exchange, err))
+	}
+}
+
+// StopConsumer 停止消费者
+func StopConsumer() {
+	if globalConsumer != nil {
+		globalConsumer.Stop()
+	}
+}

mq/internal/constant.go

@@ -1,20 +0,0 @@
-package internal
-
-import "os"
-
-const (
-	BackupPath     = "backup"
-	TimeFormat     = "20060102150405"
-	FilePathFormat = "%s/%s_%s_%09d_%04d.dat"
-)
-
-func PathExists(path string) (bool, error) {
-	_, err := os.Stat(path)
-	if err == nil {
-		return true, nil
-	}
-	if os.IsNotExist(err) {
-		return false, nil
-	}
-	return false, err
-}

mq/keyconf.go

@@ -21,7 +21,9 @@ const (
 
 // mgrsrv
 
-const ()
+const (
+	MgrAPILog = "mgr_apilog"
+)
 
 // worldsrv 消息
 
@@ -37,8 +39,6 @@ const (
 	DBCustomLog      = "db_customlog"       // 房卡场对局记录
 	DBCustomLogAward = "db_customlog_award" // 房卡场对局奖励
 	DBInvite         = "db_invite"
-	DBAPILog         = "db_apilog"
-	DBGiveLog        = "db_givelog"
 )
 
 // ranksrv 消息

mq/messagemgr.go

@@ -8,8 +8,6 @@ import (
 	"mongo.games.com/goserver/core/broker"
 	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
-
-	"mongo.games.com/game/mq/internal"
 )
 
 type HandlerFunc func(data interface{}) (err error)
@@ -50,23 +48,20 @@ func (c *MessageMgr) getName(data interface{}) string {
 	return ""
 }
 
-type RabbitMQData struct {
-	MQName string // 是队列名称也是Routing key
-	Data   interface{}
+// Write 记录消息，需要提前注册
+func (c *MessageMgr) Write(data interface{}) {
+	name := c.getName(data)
+	if name == "" {
+		name = "_null_"
+	}
+	logger.Logger.Tracef("==> RabbitMQ(%v): %#v", name, data)
+	Send(name, data)
 }
 
-func (c *MessageMgr) Send(data interface{}, name ...string) error {
-	if msg, ok := data.(*RabbitMQData); ok {
-		return Send(msg.MQName, msg.Data)
-	}
-	if len(name) > 0 && name[0] != "" {
-		return Send(name[0], data)
-	}
-	key := c.getName(data)
-	if key == "" {
-		key = "_null_"
-	}
-	return Send(key, data)
+// WriteMQData rabbitMQ消息
+func (c *MessageMgr) WriteMQData(name string, data interface{}) {
+	logger.Logger.Tracef("==> RabbitMQ(%v): %#v", name, data)
+	Send(name, data)
 }
 
 type RegisterHandlerParam struct {
@@ -95,7 +90,7 @@ func (c *MessageMgr) RegisterHandler(param *RegisterHandlerParam) {
 		F: param.Handler,
 	}
 
-	internal.RegisterSubscriber(param.Name, func(e broker.Event) (err error) {
+	RegisterSubscriber(param.Name, func(e broker.Event) (err error) {
 		msg := e.Message()
 		if msg != nil {
 			defer func() {
@@ -131,56 +126,16 @@ func RegisterMessage(param *RegisterMessageParam) {
 
 // Write 发送消息
 // 默认队列名称规则：队列前缀_消息结构体名称
-func Write(data interface{}, name ...string) error {
-	return MessageMgrSingle.Send(data, name...)
+func Write(data interface{}) {
+	MessageMgrSingle.Write(data)
+}
+
+// WriteMQData 发送消息
+func WriteMQData(name string, data interface{}) {
+	MessageMgrSingle.WriteMQData(name, data)
 }
 
 // RegisterHandler 注册消息处理函数
 func RegisterHandler(param *RegisterHandlerParam) {
 	MessageMgrSingle.RegisterHandler(param)
 }
-
-var globalConsumer *internal.RabbitMQConsumer
-
-// StartConsumer 启动消费者
-func StartConsumer(url string, exchange string, durableExchange bool, opts ...broker.Option) {
-	StopConsumer()
-	globalConsumer = internal.NewRabbitMQConsumer(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, opts...)
-	if err := globalConsumer.Start(); err != nil {
-		panic(fmt.Sprintf("RabbitMQConsumer.Start() url:%v exchange:%v err:%v", url, exchange, err))
-	}
-}
-
-// StopConsumer 停止消费者
-func StopConsumer() {
-	if globalConsumer != nil {
-		globalConsumer.Stop()
-	}
-}
-
-var globalPublisher *internal.RabbitMQPublisher
-
-// StartPublisher 启动发布者
-func StartPublisher(url string, exchange string, durableExchange bool, queueSize int, opts ...broker.Option) {
-	StopPublisher()
-	globalPublisher = internal.NewRabbitMQPublisher(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, queueSize, opts...)
-	if err := globalPublisher.Start(); err != nil {
-		panic(fmt.Sprintf("RabbitMQPublisher.Start() url:%v exchange:%v err:%v", url, exchange, err))
-	}
-}
-
-// StopPublisher 停止发布者
-func StopPublisher() {
-	if globalPublisher != nil {
-		globalPublisher.Stop()
-	}
-}
-
-func Send(topic string, msg interface{}, opts ...broker.PublishOption) (err error) {
-	if globalPublisher != nil {
-		logger.Logger.Tracef("==> RabbitMQ(%v): %#v", topic, msg)
-		return globalPublisher.Send(topic, msg, opts...)
-	}
-	logger.Logger.Errorf("RabbitMQPublisher not start!")
-	return internal.ErrClosed
-}

mq/publisher.go

@@ -1,4 +1,4 @@
-package internal
+package mq
 
 import (
 	"encoding/json"
@@ -12,14 +12,24 @@ import (
 	"mongo.games.com/goserver/core/broker"
 	"mongo.games.com/goserver/core/broker/rabbitmq"
 	"mongo.games.com/goserver/core/logger"
+
+	"mongo.games.com/game/common"
 )
 
 /*
  在发布功能上加上异步发送消息和发送失败备份到本地文件的功能
 */
 
+const (
+	BackupPath     = "backup"
+	TimeFormat     = "20060102150405"
+	FilePathFormat = "%s/%s_%s_%09d_%04d.dat"
+)
+
 var ErrClosed = errors.New("publisher is closed")
 
+var globalPublisher *RabbitMQPublisher
+
 type item struct {
 	topic string
 	msg   interface{}
@@ -111,7 +121,7 @@ func (p *RabbitMQPublisher) publishRoutine() {
 }
 
 func (p *RabbitMQPublisher) Start() (err error) {
-	if ok, _ := PathExists(BackupPath); !ok {
+	if ok, _ := common.PathExists(BackupPath); !ok {
 		err = os.MkdirAll(BackupPath, os.ModePerm)
 		if err != nil {
 			return
@@ -191,3 +201,26 @@ func (p *RabbitMQPublisher) backup(topic string, msg interface{}, err error) {
 	f.WriteString("reason:" + reason + "\n")
 	f.WriteString("data:" + string(buf) + "\n")
 }
+
+// StartPublisher 启动发布者
+func StartPublisher(url string, exchange string, durableExchange bool, queueSize int, opts ...broker.Option) {
+	StopPublisher()
+	globalPublisher = NewRabbitMQPublisher(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, queueSize, opts...)
+	if err := globalPublisher.Start(); err != nil {
+		panic(fmt.Sprintf("RabbitMQPublisher.Start() url:%v exchange:%v err:%v", url, exchange, err))
+	}
+}
+
+// StopPublisher 停止发布者
+func StopPublisher() {
+	if globalPublisher != nil {
+		globalPublisher.Stop()
+	}
+}
+
+func Send(topic string, msg interface{}, opts ...broker.PublishOption) (err error) {
+	if globalPublisher != nil {
+		return globalPublisher.Send(topic, msg, opts...)
+	}
+	return ErrClosed
+}

shell/update_deploy.bat

@@ -2,11 +2,11 @@ set deployDir="..\deploy"
 
 xcopy .\data\* %deployDir%\data\ /s /e /y
 
-for /f "tokens=*" %%a in (shell/programs.txt) do (
+for /f "tokens=*" %%a in (programs.txt) do (
     xcopy .\%%a\%%a %deployDir% /y
     del .\%%a\%%a
 )
 
-for /f "tokens=*" %%f in (shell/exclude.txt) do (
+for /f "tokens=*" %%f in (exclude.txt) do (
     if exist "%deployDir%\data\%%f" (del "%deployDir%\data\%%f")
 )

worldsrv/action_bag.go

@@ -3,6 +3,9 @@ package main
 import (
 	"fmt"
 	"math/rand"
+	"mongo.games.com/game/proto"
+	webapi_proto "mongo.games.com/game/protocol/webapi"
+	"mongo.games.com/game/webapi"
 	"strconv"
 	"time"
 
@@ -13,12 +16,8 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
-	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/bag"
-	webapiproto "mongo.games.com/game/protocol/webapi"
 	"mongo.games.com/game/srvdata"
-	"mongo.games.com/game/webapi"
 )
 
 func CSBagInfo(s *netlib.Session, packetid int, data interface{}, sid int64) error {
@@ -194,7 +193,8 @@ func CSUpBagInfo(s *netlib.Session, packetid int, data interface{}, sid int64) e
 							tp1 = model.SystemFreeGive_CoinType_Diamond
 						}
 						if !p.IsRob && tp1 >= 0 {
-							mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_CollectBox, tp1, v.ItemNum))
+							LogChannelSingleton.WriteMQData(
+								model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_CollectBox, tp1, v.ItemNum))
 						}
 					}
 				}
@@ -579,7 +579,8 @@ func CSPropExchange(s *netlib.Session, packetid int, data interface{}, sid int64
 			tp1 = model.SystemFreeGive_CoinType_Diamond
 		}
 		if !p.IsRob && tp1 >= 0 {
-			mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_CollectBoxSwap, tp1, v.ItemNum))
+			LogChannelSingleton.WriteMQData(
+				model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_CollectBoxSwap, tp1, v.ItemNum))
 		}
 	}
 
@@ -667,13 +668,13 @@ func CSDollChangeLog(s *netlib.Session, packetid int, data interface{}, sid int6
 			}
 		}
 		task.New(nil, task.CallableWrapper(func(o *basic.Object) interface{} {
-			msg := &webapiproto.ASGetDollExchangeOrder{
+			msg := &webapi_proto.ASGetDollExchangeOrder{
 				Snid:     p.SnId,
 				Platform: p.Platform,
 			}
 			buff, err := webapi.API_DollExchangeRecord(common.GetAppId(), msg)
 
-			as := &webapiproto.SAGetDollExchangeOrder{}
+			as := &webapi_proto.SAGetDollExchangeOrder{}
 			if err != nil {
 				logger.Logger.Error("API_DollExchangeRecord error:", err)
 				return nil
@@ -685,7 +686,7 @@ func CSDollChangeLog(s *netlib.Session, packetid int, data interface{}, sid int6
 			return as
 		}), task.CompleteNotifyWrapper(func(data interface{}, t task.Task) {
 			if data != nil {
-				if as := data.(*webapiproto.SAGetDollExchangeOrder); as != nil {
+				if as := data.(*webapi_proto.SAGetDollExchangeOrder); as != nil {
 					for _, v := range as.OrderList {
 						cdata := ShopMgrSington.GetExchangeData(p.Platform, v.GoodsId)
 						record := &bag.DillChangeLogInfo{

worldsrv/action_phonelottery.go

@@ -3,16 +3,13 @@ package main
 import (
 	"encoding/json"
 	"math/rand"
-
-	"mongo.games.com/goserver/core/logger"
-	"mongo.games.com/goserver/core/netlib"
-
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	player_proto "mongo.games.com/game/protocol/player"
 	"mongo.games.com/game/protocol/webapi"
 	"mongo.games.com/game/srvdata"
+	"mongo.games.com/goserver/core/logger"
+	"mongo.games.com/goserver/core/netlib"
 )
 
 type CSPhoneLotteryInfoPacketFactory struct {
@@ -198,7 +195,7 @@ func (this *CSPhoneLotteryHandler) Process(s *netlib.Session, packetid int, data
 		if err != nil {
 			return err
 		}
-		mq.Write(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 1, 0, 0, 0))
+		LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 1, 0, 0, 0))
 		for _, v := range items {
 			tp1 := int32(-1)
 			if v.ItemId == common.ItemIDCoin {
@@ -207,7 +204,8 @@ func (this *CSPhoneLotteryHandler) Process(s *netlib.Session, packetid int, data
 				tp1 = model.SystemFreeGive_CoinType_Diamond
 			}
 			if !p.IsRob && tp1 >= 0 {
-				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLottery, tp1, v.ItemNum))
+				LogChannelSingleton.WriteMQData(
+					model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLottery, tp1, v.ItemNum))
 			}
 		}
 	}

worldsrv/action_player.go

@@ -23,7 +23,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	player_proto "mongo.games.com/game/protocol/player"
 	webapi_proto "mongo.games.com/game/protocol/webapi"
@@ -1410,7 +1409,7 @@ func (this *CSPlayerRegisterHandler) Process(s *netlib.Session, packetid int, da
 			err2 := model.UpdatePlayerCoin(pi.Platform, pi.SnId, pi.Coin+int64(model.GameParamData.NewPlayerCoin), 0, 0, time.Now().Unix(), time.Now().Unix(), 0, pi.ShopID)
 			if err2 == nil {
 				if !pi.IsRob {
-					mq.Write(model.GenerateSystemFreeGive(pi.SnId, pi.Name, pi.Platform, pi.Channel, model.SystemFreeGive_GiveType_NewPlayer, model.SystemFreeGive_CoinType_Coin, int64(model.GameParamData.NewPlayerCoin)))
+					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(pi.SnId, pi.Name, pi.Platform, pi.Channel, model.SystemFreeGive_GiveType_NewPlayer, model.SystemFreeGive_CoinType_Coin, int64(model.GameParamData.NewPlayerCoin)))
 				}
 				log := model.NewCoinLogEx(&model.CoinLogParam{
 					Platform:   pi.Platform,
@@ -1428,7 +1427,7 @@ func (this *CSPlayerRegisterHandler) Process(s *netlib.Session, packetid int, da
 					Remark:     "",
 				})
 				if log != nil {
-					mq.Write(log)
+					LogChannelSingleton.WriteLog(log)
 				}
 			}
 			return player_proto.OpResultCode_OPRC_Sucess
@@ -2075,7 +2074,7 @@ func CSPlayerData(s *netlib.Session, packetid int, data interface{}, sid int64)
 			player.addLotteryCount(20)
 			player.InitLotteryStatus = true
 			player.dirty = true
-			mq.Write(model.GeneratePhoneLottery(player.SnId, player.Platform, "", 3, 20, 5, 0))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(player.SnId, player.Platform, "", 3, 20, 5, 0))
 		}
 
 		// rehold player
@@ -3083,7 +3082,7 @@ func CSUpdateAttribute(s *netlib.Session, packetId int, data interface{}, sid in
 					N:        int64(stepId),
 				}
 				b, _ := json.Marshal(d)
-				mq.Write(&model.ClientLogMysql{
+				LogChannelSingleton.WriteLog(&model.ClientLogMysql{
 					Data:     string(b),
 					Platform: p.Platform,
 					Snid:     p.SnId,

worldsrv/action_rankmatch.go

@@ -8,7 +8,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/protocol/rankmatch"
 	"mongo.games.com/game/srvdata"
 )
@@ -241,14 +240,13 @@ func CSRMAward(s *netlib.Session, packetId int, data interface{}, sid int64) err
 				case v.Id == 1: //金币
 					p.AddCoin(int64(v.Num), 0, common.GainWay_RankMatch, "system", "段位奖励")
 					if !p.IsRob {
-						mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
+						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
 							model.SystemFreeGive_GiveType_RankMatch, model.SystemFreeGive_CoinType_Coin, int64(v.Num)))
-
 					}
 				case v.Id == 2: //钻石
 					p.AddDiamond(int64(v.Num), 0, common.GainWay_RankMatch, "system", "段位奖励")
 					if !p.IsRob {
-						mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
+						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
 							model.SystemFreeGive_GiveType_RankMatch, model.SystemFreeGive_CoinType_Diamond, int64(v.Num)))
 					}
 				default:

worldsrv/action_server.go

@@ -12,7 +12,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	loginproto "mongo.games.com/game/protocol/login"
 	playerproto "mongo.games.com/game/protocol/player"
 	serverproto "mongo.games.com/game/protocol/server"
@@ -116,7 +115,7 @@ func init() {
 			}
 		}
 		//更新玩家等级排行榜
-		mq.Write(&model.PlayerLevelInfo{
+		LogChannelSingleton.WriteLog(&model.PlayerLevelInfo{
 			SnId:     p.SnId,
 			Name:     p.Name,
 			Level:    p.Level,
@@ -135,7 +134,7 @@ func init() {
 					logger.Logger.Infof("NewBankruptLogEx: snid:%v GetReturnCoin:%v coin:%v CostCoin:%v", p.SnId, msg.GetReturnCoin(), p.takeCoin, CostCoin)
 					log := model.NewBankruptLogEx(p.SnId, scene.dbGameFree.GetId(), p.CreateTime.Unix(), CostCoin, p.Platform, p.Channel, scene.gameId)
 					if log != nil {
-						mq.Write(log)
+						LogChannelSingleton.WriteLog(log)
 					}
 				}
 			}

worldsrv/action_task.go

@@ -8,7 +8,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	taskproto "mongo.games.com/game/protocol/task"
 	"mongo.games.com/game/srvdata"
 )
@@ -138,7 +137,8 @@ func SendReward(p *Player, m map[int64]int64, tp int32) {
 			tp = model.SystemFreeGive_CoinType_Diamond
 		}
 		if !p.IsRob && tp >= 0 && giveType >= 0 {
-			mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp, v.ItemNum))
+			LogChannelSingleton.WriteMQData(
+				model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp, v.ItemNum))
 		}
 	}
 }
@@ -245,7 +245,7 @@ func CSTaskReward(s *netlib.Session, packetId int, data interface{}, sid int64)
 				taskLog.ActivityType = data.GetActivityType()
 				taskLog.TaskType = data.GetTaskType()
 			}
-			mq.Write(taskLog)
+			LogChannelSingleton.WriteLog(taskLog)
 			ret.OpCode = taskproto.OpResultCode_OPRC_Success
 		}
 	}

worldsrv/action_welfare.go

@@ -12,7 +12,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	webapiproto "mongo.games.com/game/protocol/webapi"
 	"mongo.games.com/game/protocol/welfare"
@@ -579,7 +578,8 @@ func SendPermitReward(p *Player, m map[int64]int64, tp int32) {
 			tp1 = model.SystemFreeGive_CoinType_Diamond
 		}
 		if !p.IsRob && tp1 >= 0 && giveType >= 0 {
-			mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp1, v.ItemNum))
+			LogChannelSingleton.WriteMQData(
+				model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp1, v.ItemNum))
 		}
 	}
 }
@@ -1122,7 +1122,7 @@ func CSPermitExchange(s *netlib.Session, packetid int, data interface{}, sid int
 			})
 			p.WelfData.PermitExchange[msg.GetId()] = append(p.WelfData.PermitExchange[msg.GetId()], now.Unix())
 			// 兑换记录
-			mq.Write(&model.BackendPermitExchange{
+			LogChannelSingleton.WriteLog(&model.BackendPermitExchange{
 				Platform: p.Platform,
 				StartTs:  startTs,
 				SnId:     p.SnId,

worldsrv/bagmgr.go

@@ -14,7 +14,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/bag"
 	playerproto "mongo.games.com/game/protocol/player"
@@ -100,7 +99,7 @@ func init() {
 				} else {
 					permitScore = param.ItemNum
 				}
-				mq.Write(&model.BackendPermitJoin{
+				LogChannelSingleton.WriteLog(&model.BackendPermitJoin{
 					Platform: p.Platform,
 					StartTs:  PlatformMgrSingleton.GetConfig(p.Platform).PermitStartTs,
 					SnId:     p.SnId,
@@ -131,7 +130,7 @@ func init() {
 			startTs := PlatformMgrSingleton.GetConfig(p.Platform).PermitStartTs
 			if item != nil && item.ItemNum > 0 && startTs > 0 {
 				// 赛季积分排行榜
-				mq.Write(&model.PermitScore{
+				LogChannelSingleton.WriteLog(&model.PermitScore{
 					Platform: p.Platform,
 					SnId:     p.SnId,
 					Name:     p.Name,
@@ -359,7 +358,7 @@ func (this *BagMgr) CallbackAfter(ret *internal.PlayerLoadReplay) {
 					this.OnChangeFuncs(&model.ChangeItemParam{
 						SnId:         p.SnId,
 						ItemId:       v.ItemId,
-						ItemNum:      v.Count,
+						ItemNum:      num,
 						GainWay:      v.TypeId,
 						RoomConfigId: v.RoomConfigId,
 						GameId:       v.GameId,
@@ -704,7 +703,7 @@ func (this *BagMgr) AddItems(param *model.AddItemParam) (*BagInfo, bag.OpResultC
 			RoomConfigId: param.RoomConfigId,
 		})
 		if log != nil {
-			mq.Write(log)
+			LogChannelSingleton.WriteLog(log)
 			ts = log.Ts
 			newBagInfo.LogId = log.LogId.Hex()
 		}
@@ -712,7 +711,7 @@ func (this *BagMgr) AddItems(param *model.AddItemParam) (*BagInfo, bag.OpResultC
 		this.OnChangeFuncs(&model.ChangeItemParam{
 			SnId:         p.SnId,
 			ItemId:       v.ItemId,
-			ItemNum:      v.ItemNum,
+			ItemNum:      num,
 			GainWay:      param.GainWay,
 			RoomConfigId: param.RoomConfigId,
 			GameId:       param.GameId,
@@ -878,7 +877,7 @@ func (this *BagMgr) VerifyUpJybInfo(p *Player, args *model.VerifyUpJybInfoArgs)
 				if jyb.Award.Coin > 0 {
 					p.AddCoin(jyb.Award.Coin, 0, common.GainWay_ActJybAward, "system", "礼包码兑换")
 					if !p.IsRob {
-						mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Coin, int64(jyb.Award.Coin)))
+						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Coin, int64(jyb.Award.Coin)))
 					}
 					jybLog.Award = append(jybLog.Award, &model.JybItem{
 						Id:  common.ItemIDCoin,
@@ -888,7 +887,7 @@ func (this *BagMgr) VerifyUpJybInfo(p *Player, args *model.VerifyUpJybInfoArgs)
 				if jyb.Award.Diamond > 0 {
 					p.AddDiamond(jyb.Award.Diamond, 0, common.GainWay_ActJybAward, "system", "礼包码兑换")
 					if !p.IsRob {
-						mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Diamond, int64(jyb.Award.Diamond)))
+						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Diamond, int64(jyb.Award.Diamond)))
 					}
 					jybLog.Award = append(jybLog.Award, &model.JybItem{
 						Id:  common.ItemIDDiamond,
@@ -896,7 +895,7 @@ func (this *BagMgr) VerifyUpJybInfo(p *Player, args *model.VerifyUpJybInfoArgs)
 					})
 				}
 				// 领取日志
-				mq.Write(jybLog)
+				LogChannelSingleton.WriteLog(jybLog)
 				p.dirty = true
 				pack.GainItem.Coin = jyb.Award.Coin
 				pack.GainItem.Diamond = jyb.Award.Diamond

worldsrv/customroommgr.go

@@ -133,9 +133,9 @@ func (c *CustomRoomMgr) GetRoomList(plt string) []*gamehall.PrivateRoomInfo {
 }
 
 func (c *CustomRoomMgr) tryCreate(plt string, configId int32) {
-	logger.Logger.Tracef("尝试创建竞技馆系统房间 %v CloseCustomRoomCreate:%v", configId, model.GameParamData.CloseCustomRoomCreate)
+	logger.Logger.Tracef("尝试创建竞技馆系统房间 %v", configId)
 	cfg := PlatformMgrSingleton.GetConfig(plt).RoomConfigSystem[configId]
-	if model.GameParamData.CloseCustomRoomCreate || cfg == nil || cfg.GetOn() == common.Off {
+	if cfg == nil || cfg.GetOn() == common.Off {
 		return
 	}
 

worldsrv/etcd.go

@@ -14,7 +14,6 @@ import (
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/etcd"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	hallproto "mongo.games.com/game/protocol/gamehall"
 	loginproto "mongo.games.com/game/protocol/login"
 	playerproto "mongo.games.com/game/protocol/player"
@@ -266,7 +265,7 @@ func platformConfigEvent(ctx context.Context, completeKey string, isInit bool, e
 		}
 		f2 := func() {
 			if st.StartTs > 0 {
-				mq.Write(&model.BackendPermitCycle{
+				LogChannelSingleton.WriteLog(&model.BackendPermitCycle{
 					Platform: config.Platform,
 					StartTs:  st.StartTs,
 					EndTs:    st.EndTs - 1,

worldsrv/invitecode.go

@@ -5,7 +5,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 )
 
 // SaveInviteScore 保存玩家操作
@@ -16,7 +15,7 @@ func SaveInviteScore(data *model.InviteScore) {
 	}
 	cfg := PlatformMgrSingleton.GetConfig(data.Platform).ActInviteConfig
 	data.Rate = cfg.GetRates()
-	mq.Write(&model.EvtInviteMsg{
+	LogChannelSingleton.WriteLog(&model.EvtInviteMsg{
 		InviteScore:   *data,
 		RechargeScore: cfg.GetRechargeScore(),
 	})
@@ -24,7 +23,7 @@ func SaveInviteScore(data *model.InviteScore) {
 
 // CheckNewWeek 每周重置积分
 func CheckNewWeek(platform string, snid int32) {
-	mq.Write(&model.EvtInviteMsg{
+	LogChannelSingleton.WriteLog(&model.EvtInviteMsg{
 		InviteScore: model.InviteScore{
 			Platform: platform,
 			SnId:     snid,

worldsrv/logchannel.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"reflect"
+
+	"mongo.games.com/goserver/core/logger"
+
+	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
+)
+
+// LogChannelSingleton 日志记录器
+var LogChannelSingleton = &LogChannel{
+	cName: make(map[reflect.Type]string),
+}
+
+type LogChannel struct {
+	cName map[reflect.Type]string
+}
+
+func (c *LogChannel) RegisterLogCName(cname string, log interface{}) {
+	t := c.getLogType(log)
+	c.cName[t] = cname
+}
+
+func (c *LogChannel) getLogType(log interface{}) reflect.Type {
+	return reflect.Indirect(reflect.ValueOf(log)).Type()
+}
+
+func (c *LogChannel) getLogCName(log interface{}) string {
+	t := c.getLogType(log)
+	if name, exist := c.cName[t]; exist {
+		return name
+	}
+	return ""
+}
+
+// WriteLog 记录日志，需要提前注册
+func (c *LogChannel) WriteLog(log interface{}) {
+	cname := c.getLogCName(log)
+	if cname == "" {
+		cname = "_null_"
+	}
+	logger.Logger.Tracef("==> RabbitMQ(%v): %#v", cname, log)
+	mq.Send(cname, log)
+}
+
+// WriteMQData rabbitMQ消息
+func (c *LogChannel) WriteMQData(data *model.RabbitMQData) {
+	mq.Send(data.MQName, data.Data)
+}
+
+func init() {
+	LogChannelSingleton.RegisterLogCName(model.LoginLogCollName, &model.LoginLog{})
+	LogChannelSingleton.RegisterLogCName(model.CoinGiveLogCollName, &model.CoinGiveLog{})
+	LogChannelSingleton.RegisterLogCName(model.CoinLogCollName, &model.CoinLog{})
+	LogChannelSingleton.RegisterLogCName(model.ItemLogCollName, &model.ItemLog{})
+	LogChannelSingleton.RegisterLogCName(model.WelfareLogCollName, &model.WelfareLog{})
+	LogChannelSingleton.RegisterLogCName(model.OnlineLogCollName, &model.OnlineLog{})
+	LogChannelSingleton.RegisterLogCName(model.MQRankSeason, &model.PlayerRankScore{})
+	LogChannelSingleton.RegisterLogCName(model.MQRankPlayerCoin, &model.RankPlayerCoin{})
+	LogChannelSingleton.RegisterLogCName(mq.BackBankrupt, &model.BankruptLog{})
+	LogChannelSingleton.RegisterLogCName(mq.BackReliefund, &model.ReliefFundLog{})
+	LogChannelSingleton.RegisterLogCName(mq.DBInvite, &model.EvtInviteMsg{})
+	LogChannelSingleton.RegisterLogCName(model.MQRankPlayerLevel, &model.PlayerLevelInfo{})
+	LogChannelSingleton.RegisterLogCName(model.MQRankPlayerPermit, &model.PermitScore{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemPermitCycle, &model.BackendPermitCycle{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemPermitRank, &model.BackendPermitRank{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemPermitExchange, &model.BackendPermitExchange{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemPermitJoin, &model.BackendPermitJoin{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemPermitTask, &model.BackendPermitTask{})
+	LogChannelSingleton.RegisterLogCName(mq.BackClientLog, &model.ClientLogMysql{})
+	LogChannelSingleton.RegisterLogCName(mq.BackSystemJyb, &model.JybLog{})
+	LogChannelSingleton.RegisterLogCName(mq.DBVipGiftLog, &model.DbVip{})
+}

worldsrv/mq_coinlog.go

@@ -1,8 +1,13 @@
 package main
 
 import (
+	"encoding/json"
+
 	"mongo.games.com/goserver/core"
 	"mongo.games.com/goserver/core/basic"
+	"mongo.games.com/goserver/core/broker"
+	"mongo.games.com/goserver/core/broker/rabbitmq"
+	"mongo.games.com/goserver/core/logger"
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
@@ -10,39 +15,19 @@ import (
 )
 
 func init() {
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.LoginLogCollName, Data: &model.LoginLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBGiveLog, Data: &model.CoinGiveLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.CoinLogCollName, Data: &model.CoinLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.ItemLogCollName, Data: &model.ItemLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.WelfareLogCollName, Data: &model.WelfareLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.OnlineLogCollName, Data: &model.OnlineLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.MQRankSeason, Data: &model.PlayerRankScore{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.MQRankPlayerCoin, Data: &model.RankPlayerCoin{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackBankrupt, Data: &model.BankruptLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackReliefund, Data: &model.ReliefFundLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBInvite, Data: &model.EvtInviteMsg{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.MQRankPlayerLevel, Data: &model.PlayerLevelInfo{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.MQRankPlayerPermit, Data: &model.PermitScore{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemPermitCycle, Data: &model.BackendPermitCycle{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemPermitRank, Data: &model.BackendPermitRank{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemPermitExchange, Data: &model.BackendPermitExchange{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemPermitJoin, Data: &model.BackendPermitJoin{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemPermitTask, Data: &model.BackendPermitTask{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackClientLog, Data: &model.ClientLogMysql{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.BackSystemJyb, Data: &model.JybLog{}})
-	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBVipGiftLog, Data: &model.DbVip{}})
-}
+	mq.RegisterSubscriber(mq.WorldLogCoinAck, func(e broker.Event) (err error) {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
 
-func init() {
-	// 金币对账
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.WorldLogCoinAck,
-		Data: &model.CoinLog{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.CoinLog)
-			if !ok || log == nil {
+			var log model.CoinLog
+			err = json.Unmarshal(msg.Body, &log)
+			if err != nil {
 				return
 			}
+
 			//通知主线程执行后续操作
 			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
 				player := PlayerMgrSington.GetPlayerBySnId(log.SnId)
@@ -53,19 +38,26 @@ func init() {
 				return nil
 			}), true)
 			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(mq.WorldLogCoinAck), rabbitmq.DurableQueue())
 
 	// 邀请数据同步
-	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		Name: mq.WorldInviteAck,
-		Data: &model.EvtInviteAckMsg{},
-		Handler: func(data interface{}) (err error) {
-			log, ok := data.(*model.EvtInviteAckMsg)
-			if !ok || log == nil {
-				return
+	mq.RegisterSubscriber(mq.WorldInviteAck, func(e broker.Event) error {
+		msg := e.Message()
+		if msg != nil {
+			defer func() {
+				e.Ack()
+			}()
+
+			var log model.EvtInviteAckMsg
+			err := json.Unmarshal(msg.Body, &log)
+			if err != nil {
+				return err
 			}
 
+			logger.Logger.Tracef("EvtInviteAckMsg: %+v", log)
+
 			//通知主线程执行后续操作
 			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
 				player := PlayerMgrSington.GetPlayerBySnId(log.Snid)
@@ -92,7 +84,7 @@ func init() {
 				}
 				return nil
 			}), true)
-			return
-		},
-	})
+		}
+		return nil
+	}, broker.Queue(mq.WorldInviteAck), rabbitmq.DurableQueue())
 }

worldsrv/permitmgr.go

@@ -11,7 +11,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 )
 
 var PermitMgrInst = new(PermitMgr)
@@ -110,7 +109,7 @@ func (r *PermitMgr) OnDayTimer() {
 								}
 								logger.Logger.Infof("PermitMgr rank award snid:%v rank:%v", player.SnId, rank)
 								// 领奖记录
-								mq.Write(&model.BackendPermitRank{
+								LogChannelSingleton.WriteLog(&model.BackendPermitRank{
 									Platform: platform,
 									StartTs:  startTs,
 									SnId:     curPlayer.SnId,
@@ -169,7 +168,7 @@ func (r *PermitMgr) OnDayTimer() {
 				return nil
 			}), task.CompleteNotifyWrapper(func(i interface{}, t task.Task) {
 				if se.StartTs > 0 {
-					mq.Write(&model.BackendPermitCycle{
+					LogChannelSingleton.WriteLog(&model.BackendPermitCycle{
 						Platform: platform,
 						StartTs:  se.StartTs,
 						EndTs:    se.EndTs - 1,

worldsrv/player.go

@@ -25,7 +25,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	"mongo.games.com/game/protocol/bag"
 	hallproto "mongo.games.com/game/protocol/gamehall"
@@ -347,7 +346,7 @@ func (this *Player) OnLogined() {
 			this.addLotteryCount(20)
 			this.InitLotteryStatus = true
 			this.dirty = true
-			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 20, 5, 0))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 20, 5, 0))
 		}
 
 		//登录日志
@@ -356,7 +355,7 @@ func (this *Player) OnLogined() {
 		if logState != nil {
 			clog = logState.clog
 		}
-		mq.Write(model.NewLoginLog(this.SnId, common.LoginLogTypeLogin, this.Tel, this.Ip,
+		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeLogin, this.Tel, this.Ip,
 			this.Platform, this.Channel, this.BeUnderAgentCode, this.PackageID, this.City, clog, this.GetTotalCoin(), 0, 0,
 			this.DeviceName, this.PackageName, this.AppVersion, this.BuildVersion, this.AppChannel, this.ChannelId))
 
@@ -429,7 +428,7 @@ func (this *Player) OnRehold() {
 		if logState != nil {
 			clog = logState.clog
 		}
-		mq.Write(model.NewLoginLog(this.SnId, common.LoginLogTypeRehold, this.Tel, this.Ip,
+		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeRehold, this.Tel, this.Ip,
 			this.Platform, this.Channel, this.BeUnderAgentCode, this.PackageID, this.City, clog, this.GetTotalCoin(),
 			gameid, 0, this.DeviceName, this.PackageName, this.AppVersion, this.BuildVersion, this.AppChannel, this.ChannelId))
 
@@ -1118,8 +1117,9 @@ func (this *Player) GetMessageAttach(id string) {
 										tp = model.SystemFreeGive_CoinType_Diamond
 									}
 									if tp != -1 {
-										mq.Write(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel,
-											model.SystemFreeGive_GiveType_TaskPermitRank, tp, int64(msg.Params[i+1])))
+										LogChannelSingleton.WriteMQData(
+											model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel,
+												model.SystemFreeGive_GiveType_TaskPermitRank, tp, int64(msg.Params[i+1])))
 									}
 								}
 							}
@@ -1202,7 +1202,7 @@ func (this *Player) GetMessageAttach(id string) {
 						this.AddPayCoinLog(msg.Coin, model.PayCoinLogType_Coin, "mail")
 						if msg.Oper == 0 { //系统赠送
 							if !this.IsRob {
-								mq.Write(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
+								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
 									model.SystemFreeGive_CoinType_Coin, int64(msg.Coin)))
 							}
 						}
@@ -1219,7 +1219,7 @@ func (this *Player) GetMessageAttach(id string) {
 						this.AddDiamond(msg.Diamond, 0, gainWay, msg.Id.Hex(), remark)
 						if msg.Oper == 0 { //系统赠送
 							if !this.IsRob {
-								mq.Write(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
+								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
 									model.SystemFreeGive_CoinType_Diamond, int64(msg.Diamond)))
 							}
 						}
@@ -1281,7 +1281,7 @@ func (this *Player) DropLine() {
 		if this.scene != nil && this.scene.gameSess != nil {
 			gameid = this.scene.gameId
 		}
-		mq.Write(model.NewLoginLog(this.SnId, common.LoginLogTypeDrop, this.Tel, this.Ip,
+		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeDrop, this.Tel, this.Ip,
 			this.Platform, this.Channel, this.BeUnderAgentCode, this.PackageID, this.City, clog, this.GetTotalCoin(),
 			gameid, this.LastGameId, this.DeviceName, this.PackageName, this.AppVersion, this.BuildVersion, this.AppChannel, this.ChannelId))
 
@@ -1406,7 +1406,7 @@ func (this *Player) OnLogouted() {
 	}
 	//排除掉机器人
 	if !this.IsRob {
-		mq.Write(model.NewLoginLog(this.SnId, common.LoginLogTypeLogout, this.Tel, this.Ip,
+		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeLogout, this.Tel, this.Ip,
 			this.Platform, this.Channel, this.BeUnderAgentCode, this.PackageID, this.City, clog, this.GetTotalCoin(),
 			0, 0, this.DeviceName, this.PackageName, this.AppVersion, this.BuildVersion, this.AppChannel, this.ChannelId))
 		this.OnlineLogLogout()
@@ -1712,7 +1712,7 @@ func (this *Player) AddDiamond(num, add int64, gainWay int32, oper, remark strin
 			Remark:     remark,
 		})
 		if log != nil {
-			mq.Write(log)
+			LogChannelSingleton.WriteLog(log)
 		}
 	}
 }
@@ -1767,7 +1767,7 @@ func (this *Player) AddCoin(num, add int64, gainWay int32, oper, remark string)
 				Remark:     remark,
 			})
 			if log != nil {
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 			}
 		}
 	}
@@ -1822,7 +1822,7 @@ func (this *Player) AddCoinAsync(num, add int64, gainWay int32, oper, remark str
 				Remark:     remark,
 			})
 			if log != nil {
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 			}
 		}
 	}
@@ -2033,22 +2033,7 @@ func (this *Player) ResetPermit() {
 		// 清理数据
 		bag := BagMgrSingleton.GetBagInfo(this.SnId)
 		if bag != nil {
-			if bag.BagItem[common.ItemIDPermit] != nil {
-				BagMgrSingleton.AddItems(&model.AddItemParam{
-					Platform: this.Platform,
-					SnId:     this.SnId,
-					Change: []*model.Item{
-						{
-							ItemId:  common.ItemIDPermit,
-							ItemNum: -bag.BagItem[common.ItemIDPermit].ItemNum,
-						},
-					},
-					GainWay:  common.GainWayPermitReset,
-					Operator: "system",
-					Remark:   "赛季积分清理",
-				})
-			}
-
+			delete(bag.BagItem, common.ItemIDPermit)
 			if model.GameParamData.PermitInitScore > 0 {
 				bagInfo := BagMgrSingleton.GetBagInfo(this.SnId)
 				if bagInfo != nil {
@@ -2057,21 +2042,6 @@ func (this *Player) ResetPermit() {
 						ItemNum:    model.GameParamData.PermitInitScore,
 						ObtainTime: time.Now().Unix(),
 					}
-
-					BagMgrSingleton.AddItems(&model.AddItemParam{
-						Platform: this.Platform,
-						SnId:     this.SnId,
-						Change: []*model.Item{
-							{
-								ItemId:  common.ItemIDPermit,
-								ItemNum: model.GameParamData.PermitInitScore,
-							},
-						},
-						GainWay:  common.GainWayPermitReset,
-						Operator: "system",
-						Remark:   "初始赛季积分",
-					})
-
 					this.Permit = time.Now()
 				}
 			}
@@ -2938,7 +2908,7 @@ func (this *Player) ReportLoginEvent() {
 				this.UpgradeTime = this.CreateTime
 			}
 		}
-		mq.Write(model.GenerateLogin(model.CreatePlayerLoginEvent(this.SnId,
+		LogChannelSingleton.WriteMQData(model.GenerateLogin(model.CreatePlayerLoginEvent(this.SnId,
 			this.Channel, this.BeUnderAgentCode, this.Platform, this.City, this.DeviceOS, this.Ip,
 			this.CreateTime, this.UpgradeTime, isBindPhone, this.TelephonePromoter, this.DeviceId, this.ChannelId)))
 		//登录通知
@@ -3285,7 +3255,7 @@ func (this *Player) TryRetrieveLostGameCoin(sceneid int) {
 	logProbe.RestCount = 0 //this.Coin
 	logProbe.SeqNo = this.GameCoinTs
 	logProbe.RoomId = int32(sceneid)
-	mq.Write(logProbe)
+	LogChannelSingleton.WriteLog(logProbe)
 	this.SendDiffData()
 }
 
@@ -3514,7 +3484,7 @@ func (this *Player) VIPDraw(id, vip int32) {
 			var item []model.ItemInfo
 			item = append(item, model.ItemInfo{ItemId: 1, ItemNum: money})
 			log := model.NewDbVip(this.Platform, this.SnId, this.VIP, 0, 0, 0, item, 0, "Vip每日礼包")
-			mq.Write(log)
+			LogChannelSingleton.WriteLog(log)
 
 		case 1:
 			var items []*model.Item
@@ -3538,7 +3508,7 @@ func (this *Player) VIPDraw(id, vip int32) {
 			logger.Logger.Tracef("玩家领取VIP等级礼包成功！snid = %v,Vip = %v,items %#v", this.SnId, this.VIP, items)
 			//VIP礼包统计数据
 			log := model.NewDbVip(this.Platform, this.SnId, this.VIP, 9, 0, 0, itemInfo, 0, "Vip等级礼包")
-			mq.Write(log)
+			LogChannelSingleton.WriteLog(log)
 		}
 	}
 	send()
@@ -3683,7 +3653,7 @@ func (this *Player) DoShopInfo(info *model.DbShop, isLogin bool) {
 		// 通行证
 		if info.PageId == ShopPagePermit {
 			this.Permit = info.CreateTs.Local()
-			mq.Write(&model.BackendPermitJoin{
+			LogChannelSingleton.WriteLog(&model.BackendPermitJoin{
 				Platform: this.Platform,
 				StartTs:  PlatformMgrSingleton.GetConfig(this.Platform).PermitStartTs,
 				SnId:     this.SnId,
@@ -3797,7 +3767,7 @@ func (this *Player) OnlineLogLogin() {
 	}
 	if this.onlineLog == nil {
 		this.onlineLog = model.NewOnlineLog(bson.NewObjectId(), this.SnId, common.LoginLogTypeLogin, this.Platform)
-		mq.Write(*this.onlineLog)
+		LogChannelSingleton.WriteLog(*this.onlineLog)
 	}
 }
 
@@ -3807,7 +3777,7 @@ func (this *Player) OnlineLogRehold() {
 	}
 	if this.onlineLog == nil {
 		this.onlineLog = model.NewOnlineLog(bson.NewObjectId(), this.SnId, common.LoginLogTypeRehold, this.Platform)
-		mq.Write(*this.onlineLog)
+		LogChannelSingleton.WriteLog(*this.onlineLog)
 	}
 }
 
@@ -3818,7 +3788,7 @@ func (this *Player) OnlineLogDrop() {
 	if this.onlineLog != nil && this.onlineLog.OfflineTs == 0 {
 		this.onlineLog.OfflineType = common.LoginLogTypeDrop
 		this.onlineLog.OfflineTs = time.Now().Unix()
-		mq.Write(*this.onlineLog)
+		LogChannelSingleton.WriteLog(*this.onlineLog)
 		this.onlineLog = nil
 	}
 }
@@ -3830,7 +3800,7 @@ func (this *Player) OnlineLogLogout() {
 	if this.onlineLog != nil && this.onlineLog.OfflineTs == 0 {
 		this.onlineLog.OfflineType = common.LoginLogTypeLogout
 		this.onlineLog.OfflineTs = time.Now().Unix()
-		mq.Write(*this.onlineLog)
+		LogChannelSingleton.WriteLog(*this.onlineLog)
 		this.onlineLog = nil
 	}
 }
@@ -3909,7 +3879,7 @@ func (this *Player) SendRankSeason() {
 }
 
 func (this *Player) SendPlayerCoin() {
-	mq.Write(&model.RankPlayerCoin{
+	LogChannelSingleton.WriteLog(&model.RankPlayerCoin{
 		Platform: this.Platform,
 		SnId:     this.SnId,
 		Name:     this.Name,
@@ -3999,7 +3969,7 @@ func (this *Player) UpdatePlayerVipBag(shopId int32) {
 			item = append(item, model.ItemInfo{ItemId: shopInfo.Type, ItemNum: shopInfo.Amount})
 			log := model.NewDbVip(this.Platform, this.SnId, shopInfo.VipLevel, shopInfo.Type, shopInfo.ConstType,
 				int64(shopInfo.CostArea[0]), item, shopInfo.Id, shopInfo.Name)
-			mq.Write(log)
+			LogChannelSingleton.WriteLog(log)
 		}
 		//判断是否是礼包周卡
 		if shopInfo.Page == ShopPageGift {
@@ -4223,7 +4193,7 @@ func (this *Player) PhoneLotteryTask(taskId int32, num int64) {
 			count := this.WelfData.PhoneLotteryTask[taskId].N / 20
 			this.addLotteryCount(int32(count))
 			this.WelfData.PhoneLotteryTask[taskId].N = this.WelfData.PhoneLotteryTask[taskId].N % 20
-			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 4, 0))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 4, 0))
 		}
 
 	case common.TaskTypeWinOrLose:
@@ -4232,7 +4202,7 @@ func (this *Player) PhoneLotteryTask(taskId int32, num int64) {
 			count := this.WelfData.PhoneLotteryTask[taskId].N / 200000000
 			this.addLotteryCount(int32(count))
 			this.WelfData.PhoneLotteryTask[taskId].N = this.WelfData.PhoneLotteryTask[taskId].N % 200000000
-			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 3, 0))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 3, 0))
 		}
 
 	case common.TaskTypeTienlenCount:
@@ -4241,11 +4211,11 @@ func (this *Player) PhoneLotteryTask(taskId int32, num int64) {
 			count := this.WelfData.PhoneLotteryTask[taskId].N / 20
 			this.addLotteryCount(int32(count))
 			this.WelfData.PhoneLotteryTask[taskId].N = this.WelfData.PhoneLotteryTask[taskId].N % 20
-			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 2, 0))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 2, 0))
 		}
 	case common.TaskTypeFirstLogin:
 		this.addLotteryCount(5)
-		mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 5, 1, 0))
+		LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 5, 1, 0))
 	default:
 		logger.Logger.Errorf("手机抽奖任务未找到对应的任务ID: %d", taskId)
 		return

worldsrv/playeronline.go

@@ -6,7 +6,6 @@ import (
 	"mongo.games.com/goserver/core/module"
 
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 )
 
 var PlayerOnlineSington = &PlayerOnlineEvent{
@@ -59,7 +58,7 @@ func (p *PlayerOnlineEvent) Update() {
 
 here:
 	p.OnlineCh = onlineCh
-	mq.Write(model.GenerateOnline(p.OnlineCh))
+	LogChannelSingleton.WriteMQData(model.GenerateOnline(p.OnlineCh))
 }
 
 func (p *PlayerOnlineEvent) Shutdown() {

worldsrv/rankmatch.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"mongo.games.com/game/common"
 	"sort"
 	"strings"
 	"time"
@@ -15,10 +16,8 @@ import (
 	"mongo.games.com/goserver/core/module"
 	"mongo.games.com/goserver/core/task"
 
-	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
-	hallproto "mongo.games.com/game/protocol/gamehall"
+	hall_proto "mongo.games.com/game/protocol/gamehall"
 	"mongo.games.com/game/protocol/rankmatch"
 	"mongo.games.com/game/srvdata"
 	"mongo.games.com/game/worldsrv/internal"
@@ -214,9 +213,9 @@ here:
 	p := PlayerMgrSington.GetPlayerBySnId(this.SnId)
 	if p != nil {
 		if has {
-			p.SendShowRed(hallproto.ShowRedCode_RankReward, rankType, 1)
+			p.SendShowRed(hall_proto.ShowRedCode_RankReward, rankType, 1)
 		} else {
-			p.SendShowRed(hallproto.ShowRedCode_RankReward, rankType, 0)
+			p.SendShowRed(hall_proto.ShowRedCode_RankReward, rankType, 0)
 		}
 	}
 }
@@ -242,7 +241,7 @@ func (this *PlayerRankSeason) rankLog(rankType int32) {
 		UpdateTs: time.Now().Unix(),
 		ModId:    p.PlayerData.GetRoleId(),
 	}
-	mq.Write(log)
+	LogChannelSingleton.WriteLog(log)
 }
 
 type RankMatchMgr struct {
@@ -473,7 +472,7 @@ func (r *RankMatchMgr) UpdateRobotSeason(platform string, snid int32, rankType i
 		UpdateTs: time.Now().Unix(),
 		ModId:    roleId,
 	}
-	mq.Write(log)
+	LogChannelSingleton.WriteLog(log)
 }
 
 func (r *RankMatchMgr) SetPlayerSeason(config *PlayerRankSeason) {

worldsrv/shopmgr.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/rand"
+	"mongo.games.com/game/protocol/bag"
 	"slices"
 	"strconv"
 	"time"
@@ -16,9 +17,7 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
-	"mongo.games.com/game/protocol/bag"
 	hall_proto "mongo.games.com/game/protocol/gamehall"
 	"mongo.games.com/game/protocol/shop"
 	webapi_proto "mongo.games.com/game/protocol/webapi"
@@ -596,7 +595,7 @@ func (this *ShopMgr) createPhoneScore(p *Player, shopInfo *model.ShopInfo, costN
 		})
 		jsonData, err := json.Marshal(items)
 		if err == nil {
-			mq.Write(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 2, 0, 0, int(costNum)))
+			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 2, 0, 0, int(costNum)))
 		}
 		for _, v := range items {
 			tp1 := int32(-1)
@@ -606,7 +605,8 @@ func (this *ShopMgr) createPhoneScore(p *Player, shopInfo *model.ShopInfo, costN
 				tp1 = model.SystemFreeGive_CoinType_Diamond
 			}
 			if !p.IsRob && tp1 >= 0 {
-				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLotterySwap, tp1, v.ItemNum))
+				LogChannelSingleton.WriteMQData(
+					model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLotterySwap, tp1, v.ItemNum))
 			}
 		}
 	}
@@ -749,7 +749,7 @@ func (this *ShopMgr) GainShop(shopInfo *model.ShopInfo, p *Player, vipShopId, po
 		p.AddCoin(addTotal, 0, common.GainWay_Shop_Buy, "system", shopName)
 		if shopInfo.Ad > 0 { //观看广告
 			if !p.IsRob {
-				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Coin, addTotal))
+				LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Coin, addTotal))
 			}
 		}
 		// 记录钻石兑换金币的金币数量，个人水池调控使用
@@ -766,7 +766,7 @@ func (this *ShopMgr) GainShop(shopInfo *model.ShopInfo, p *Player, vipShopId, po
 		p.AddDiamond(addTotal, 0, common.GainWay_Shop_Buy, "system", shopName)
 		if shopInfo.Ad > 0 { //观看广告
 			if !p.IsRob {
-				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Diamond, addTotal))
+				LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Diamond, addTotal))
 			}
 		}
 	default:

worldsrv/task_login.go

@@ -17,7 +17,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	"mongo.games.com/game/proto"
 	login_proto "mongo.games.com/game/protocol/login"
 	server_proto "mongo.games.com/game/protocol/server"
@@ -134,7 +133,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 			if err == nil {
 				// 新号赠送日志
 				if !playerData.IsRob {
-					mq.Write(model.GenerateSystemFreeGive(
+					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
 						playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 						model.SystemFreeGive_GiveType_NewPlayer, model.SystemFreeGive_CoinType_Coin,
 						int64(model.GameParamData.NewPlayerCoin)))
@@ -156,7 +155,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 					Remark:     "",
 				})
 				if log != nil {
-					mq.Write(log)
+					LogChannelSingleton.WriteLog(log)
 				}
 			}
 
@@ -168,7 +167,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 						// 金币
 						if coin > 0 {
 							if !playerData.IsRob {
-								mq.Write(model.GenerateSystemFreeGive(
+								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
 									playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 									model.SystemFreeGive_GiveType_BindTel, model.SystemFreeGive_CoinType_Coin, coin))
 							}
@@ -189,13 +188,13 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 								Remark:     "",
 							})
 							if log != nil {
-								mq.Write(log)
+								LogChannelSingleton.WriteLog(log)
 							}
 						}
 						// 钻石
 						if diamond > 0 {
 							if !playerData.IsRob {
-								mq.Write(model.GenerateSystemFreeGive(
+								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
 									playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 									model.SystemFreeGive_GiveType_BindTel, model.SystemFreeGive_CoinType_Diamond, diamond))
 							}
@@ -216,7 +215,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 								Remark:     "",
 							})
 							if log != nil {
-								mq.Write(log)
+								LogChannelSingleton.WriteLog(log)
 							}
 						}
 						//todo道具

worldsrv/transact_minigameaddcoin.go

@@ -55,7 +55,7 @@ package main
 //			log := model.NewCoinLogEx(p.SnId, ctx.Coin, p.Coin, p.SafeBoxCoin, p.Ver, ctx.GainWay, 0,
 //				ctx.Oper, ctx.Remark, p.Platform, p.Channel, p.BeUnderAgentCode, 0, p.PackageID, 0)
 //			if log != nil {
-//				mq.Write(log)
+//				LogChannelSingleton.WriteLog(log)
 //			}
 //		}
 //		tNode.TransRep.RetFiels = p.Coin

worldsrv/welfmgr.go

@@ -10,7 +10,6 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
-	"mongo.games.com/game/mq"
 	hall_proto "mongo.games.com/game/protocol/gamehall"
 	player_proto "mongo.games.com/game/protocol/player"
 	"mongo.games.com/game/protocol/shop"
@@ -110,7 +109,7 @@ func (this *WelfareMgr) GetReliefFund(p *Player, isVideo bool) {
 
 			log := model.NewReliefFundLogEx(p.SnId, getType, model.SystemFreeGive_CoinType_Coin, coin, p.CreateTime.Unix(), p.Platform, p.Channel)
 			if log != nil {
-				mq.Write(log)
+				LogChannelSingleton.WriteLog(log)
 				logger.Logger.Tracef("NewReliefFundLogEx WriteLog snid: %v Coin:%v", p.SnId, pack.Coin)
 			}
 		}
@@ -557,14 +556,14 @@ func DrawWelfareDate(dates []*webapi_proto.WelfareDate, p *Player, gainWay int32
 			p.AddCoin(coin, add, gainWay, oper, remark)
 			if giveType != -1 {
 				if !p.IsRob {
-					mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Coin, int64(coin)))
+					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Coin, int64(coin)))
 				}
 			}
 		case 2: //钻石
 			p.AddDiamond(int64(v.Grade), 0, gainWay, oper, remark)
 			if giveType != -1 {
 				if !p.IsRob {
-					mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Diamond, int64(v.Grade)))
+					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Diamond, int64(v.Grade)))
 				}
 			}
 		case 3: //道具
@@ -1155,7 +1154,7 @@ func (this *WelfareMgr) RecordWelfareLog(p *Player, num, takenum, discount int64
 
 	if log != nil {
 		//logger.Logger.Trace("RecordItemLog 开始记录 道具操作")
-		mq.Write(log)
+		LogChannelSingleton.WriteLog(log)
 	}
 }