rabbitmq优化

dbproxy/main.go

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

dbproxy/mq/c_apilog.go

@@ -1,36 +1,25 @@
 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.RegisterSubscriber(mq.MgrAPILog, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: mq.DBAPILog,
-		if msg != nil {
+		Data: model.APILog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.APILog)
-			}()
+			if !ok {
-
+				return nil
-			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
+			return nil
-		}
+		},
-		return nil
+	})
-	}, broker.Queue(mq.MgrAPILog), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_coingivelog.go

@@ -1,36 +1,25 @@
 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.RegisterSubscriber(model.CoinGiveLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: mq.DBGiveLog,
-		if msg != nil {
+		Data: model.CoinGiveLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.CoinGiveLog)
-			}()
+			if !ok {
-
-			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,30 +1,20 @@
 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.RegisterSubscriber(model.CoinLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.CoinLogCollName,
-		if msg != nil {
+		Data: &model.CoinLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.CoinLog)
-			}()
+			if !ok {
-
-			var log model.CoinLog
-			err = json.Unmarshal(msg.Body, &log)
-			if err != nil {
 				return
 			}
-
 			if log.Count == 0 { //玩家冲账探针
 				mq.Send(mq.WorldLogCoinAck, log)
 			} else {
@@ -40,7 +30,6 @@ func init() {
 				}
 			}
 			return
-		}
+		},
-		return nil
+	})
-	}, broker.Queue(model.CoinLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_customlog.go

@@ -7,6 +7,12 @@ 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,32 +1,22 @@
 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.RegisterSubscriber(model.FriendRecordLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.FriendRecordLogCollName,
-		if msg != nil {
+		Data: &model.FriendRecord{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.FriendRecord)
-			}()
+			if !ok {
-
-			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,10 +1,6 @@
 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"
@@ -13,19 +9,14 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(model.GameDetailedLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.GameDetailedLogCollName,
-		if msg != nil {
+		Data: &model.GameDetailedLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.GameDetailedLog)
-			}()
+			if !ok {
-
-			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)
@@ -34,7 +25,6 @@ func init() {
 				}
 			}
 			return
-		}
+		},
-		return nil
+	})
-	}, broker.Queue(model.GameDetailedLogCollName), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_gameplayerlistlog.go

@@ -1,36 +1,25 @@
 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.RegisterSubscriber(model.GamePlayerListLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.GamePlayerListLogCollName,
-		if msg != nil {
+		Data: &model.GamePlayerListLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.GamePlayerListLog)
-			}()
+			if !ok {
-
-			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,14 +2,11 @@ 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"
@@ -20,21 +17,14 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(mq.DBInvite, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: mq.DBInvite,
-		if msg != nil {
+		Data: &model.EvtInviteMsg{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.EvtInviteMsg)
-			}()
+			if !ok {
-
-			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)
@@ -303,8 +293,7 @@ func init() {
 				}
 				mq.Send(mq.WorldInviteAck, msg)
 			}
-
+			return
-		}
+		},
-		return nil
+	})
-	}, broker.Queue(mq.DBInvite), broker.DisableAutoAck(), rabbitmq.DurableQueue())
 }

dbproxy/mq/c_itemlog.go

@@ -1,36 +1,25 @@
 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.RegisterSubscriber(model.ItemLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.ItemLogCollName,
-		if msg != nil {
+		Data: &model.ItemLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.ItemLog)
-			}()
+			if !ok {
-
-			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,38 +1,27 @@
 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.RegisterSubscriber(model.LoginLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.LoginLogCollName,
-		if msg != nil {
+		Data: &model.LoginLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.LoginLog)
-			}()
+			if !ok {
-
-			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,37 +1,26 @@
 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.RegisterSubscriber(model.OnlineLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.OnlineLogCollName,
-		if msg != nil {
+		Data: &model.OnlineLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.OnlineLog)
-			}()
+			if !ok {
-
-			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,12 +1,6 @@
 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"
@@ -16,117 +10,70 @@ import (
 
 func init() {
 	// 排位排行榜
-	mq.RegisterSubscriber(model.MQRankSeason, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.MQRankSeason,
-		if msg != nil {
+		Data: &model.PlayerRankScore{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.PlayerRankScore)
-			}()
+			if !ok {
-
-			var log model.PlayerRankScore
-			err = json.Unmarshal(msg.Body, &log)
-			if err != nil {
 				return
 			}
-
+			err = svc.RankSeasonUpsert(log)
-			logger.Logger.Tracef("SubscriberPlayerRankScore: %+v", log)
+			if err != nil {
-
+				logger.Logger.Errorf("RankSeasonUpsert err: %v", err)
-			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.RegisterSubscriber(model.MQRankPlayerCoin, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.MQRankPlayerCoin,
-		if msg != nil {
+		Data: &model.RankPlayerCoin{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.RankPlayerCoin)
-			}()
+			if !ok {
-
-			var log model.RankPlayerCoin
-			err = json.Unmarshal(msg.Body, &log)
-			if err != nil {
 				return
 			}
-
+			err = svc.RankPlayerCoinUpsert(log)
-			logger.Logger.Tracef("SubscriberRankPlayerCoin: %+v", log)
+			if err != nil {
-
+				logger.Logger.Errorf("RankPlayerCoinUpsert err: %v", err)
-			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.RegisterSubscriber(model.MQRankPlayerLevel, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.MQRankPlayerLevel,
-		if msg != nil {
+		Data: &model.PlayerLevelInfo{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.PlayerLevelInfo)
-			}()
+			if !ok {
-
-			var log model.PlayerLevelInfo
-			err = json.Unmarshal(msg.Body, &log)
-			if err != nil {
 				return
 			}
-
+			err = svc.RankPlayerLevelUpsert(log)
-			logger.Logger.Tracef("SubscriberRankPlayerLevel: %+v", log)
+			if err != nil {
-
+				logger.Logger.Errorf("RankPlayerLevelUpsert err: %v", err)
-			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) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.MQRankPlayerPermit,
-		if msg != nil {
+		Data: &model.PermitScore{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.PermitScore)
-			}()
+			if !ok {
-
-			var log model.PermitScore
-			err = json.Unmarshal(msg.Body, &log)
-			if err != nil {
 				return
 			}
-
+			err = svc.RankPlayerPermitUpsert(log)
-			logger.Logger.Tracef("SubscriberRankPlayerPermit: %+v", log)
+			if err != nil {
-
+				logger.Logger.Errorf("RankPlayerPermitUpsert err: %v", err)
-			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,10 +1,7 @@
 package mq
 
 import (
-	"encoding/json"
+	"mongo.games.com/goserver/core/logger"
-
-	"mongo.games.com/goserver/core/broker"
-	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -12,25 +9,22 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(model.SceneCoinLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.SceneCoinLogCollName,
-		if msg != nil {
+		Data: &model.SceneCoinLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.SceneCoinLog)
-			}()
+			if !ok {
-
-			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,10 +1,7 @@
 package mq
 
 import (
-	"encoding/json"
+	"mongo.games.com/goserver/core/logger"
-
-	"mongo.games.com/goserver/core/broker"
-	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -12,25 +9,22 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(mq.DBVipGiftLog, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: mq.DBVipGiftLog,
-		if msg != nil {
+		Data: &model.DbVip{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.DbVip)
-			}()
+			if !ok {
-
-			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,10 +1,7 @@
 package mq
 
 import (
-	"encoding/json"
+	"mongo.games.com/goserver/core/logger"
-
-	"mongo.games.com/goserver/core/broker"
-	"mongo.games.com/goserver/core/broker/rabbitmq"
 
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
@@ -12,25 +9,22 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(model.WelfareLogCollName, func(e broker.Event) (err error) {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: model.WelfareLogCollName,
-		if msg != nil {
+		Data: &model.WelfareLog{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.WelfareLog)
-			}()
+			if !ok {
-
-			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

@@ -1,19 +0,0 @@
-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

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

gamesrv/base/replay_recorder.go

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

gamesrv/base/scene.go

@@ -16,6 +16,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/gamehall"
 	"mongo.games.com/game/protocol/player"
@@ -1435,7 +1436,7 @@ func (this *Scene) SaveGameDetailedLog(logid string, gamedetailednote string, ga
 							if this.IsCustom() {
 								log.CycleId = this.CycleID
 							}
-							LogChannelSingleton.WriteLog(log)
+							mq.Write(log)
 						}
 					}
 				}
@@ -1454,7 +1455,7 @@ func (this *Scene) SaveGameDetailedLog(logid string, gamedetailednote string, ga
 					}
 					newLog := new(model.GameDetailedLog)
 					*newLog = *log
-					LogChannelSingleton.WriteLog(log)
+					mq.Write(log)
 				}
 			}
 		}
@@ -1514,7 +1515,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 {
-		LogChannelSingleton.WriteLog(log)
+		mq.Write(log)
 	}
 }
 
@@ -1563,7 +1564,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 {
-				LogChannelSingleton.WriteLog(log)
+				mq.Write(log)
 			}
 		}
 	}

gamesrv/main.go

@@ -12,6 +12,7 @@ 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

@@ -0,0 +1,19 @@
+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,6 +16,7 @@ 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"
@@ -2163,5 +2164,5 @@ func (this *TienLenSceneData) SaveCustomLog() {
 			LogId: this.RoundLogId[k],
 		})
 	}
-	base.LogChannelSingleton.WriteLog(log)
+	mq.Write(log)
 }

gamesrv/tienlen/scenepolicy_tienlen.go

@@ -14,6 +14,7 @@ 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"
@@ -2639,7 +2640,7 @@ func (this *SceneBilledStateTienLen) OnEnter(s *base.Scene) {
 								GameId:     int64(sceneEx.GameId),
 								GameFreeId: int64(sceneEx.GetGameFreeId()),
 							})
-							base.LogChannelSingleton.WriteLog(&model.CustomLogAward{
+							mq.Write(&model.CustomLogAward{
 								Platform: p.Platform,
 								CycleId:  sceneEx.CycleID,
 								SnId:     p.SnId,

gatesrv/clientlog.go

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

gatesrv/logchannel.go

@@ -1,46 +0,0 @@
-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,6 +6,7 @@ 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

@@ -0,0 +1,10 @@
+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

@@ -1,57 +0,0 @@
-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,6 +20,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/webapi"
 )
@@ -121,7 +122,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)
-		LogChannelSington.WriteLog(log)
+		mq.Write(log)
 	}
 	return
 }

mgrsrv/api/webapi_worldsrv.go

@@ -21,6 +21,7 @@ import (
 
 	"mongo.games.com/game/common"
 	"mongo.games.com/game/model"
+	"mongo.games.com/game/mq"
 )
 
 // API
@@ -133,7 +134,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)
-		LogChannelSington.WriteLog(log)
+		mq.Write(log)
 	}
 	return
 }

mgrsrv/main.go

@@ -9,6 +9,7 @@ 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

@@ -0,0 +1,10 @@
+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/rabbit_mq.go

@@ -8,7 +8,7 @@ import (
 )
 
 // GenerateOnline 在线统计
-func GenerateOnline(online map[string]map[string]int) *RabbitMQData {
+func GenerateOnline(online map[string]map[string]int) *mq.RabbitMQData {
 	params := make(map[string]interface{})
 	params["Online"] = online
 	params["Time"] = time.Now().Unix()
@@ -16,17 +16,12 @@ func GenerateOnline(online map[string]map[string]int) *RabbitMQData {
 }
 
 // GenerateLogin 玩家登陆事件
-func GenerateLogin(o *PlayerLoginEvent) *RabbitMQData {
+func GenerateLogin(o *PlayerLoginEvent) *mq.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) *RabbitMQData {
+func GenerateSystemFreeGive(snid int32, name, platform, channel string, givetype, cointype int32, count int64) *mq.RabbitMQData {
 	params := make(map[string]string)
 	params["snid"] = strconv.Itoa(int(snid))
 	params["Channel"] = channel
@@ -40,7 +35,7 @@ func GenerateSystemFreeGive(snid int32, name, platform, channel string, givetype
 }
 
 // GeneratePhoneLottery 手机抽奖统计
-func GeneratePhoneLottery(snid int32, platform string, items string, lotteryType, count, countType, score int) *RabbitMQData {
+func GeneratePhoneLottery(snid int32, platform string, items string, lotteryType, count, countType, score int) *mq.RabbitMQData {
 	params := make(map[string]string)
 	params["Snid"] = strconv.Itoa(int(snid))
 	params["Platform"] = platform

model/rabbitmqdata.go

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

mq/export.go

@@ -8,6 +8,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/mq/internal"
 )
 
 type HandlerFunc func(data interface{}) (err error)
@@ -48,20 +50,23 @@ func (c *MessageMgr) getName(data interface{}) string {
 	return ""
 }
 
-// Write 记录消息，需要提前注册
+type RabbitMQData struct {
-func (c *MessageMgr) Write(data interface{}) {
+	MQName string // 是队列名称也是Routing key
-	name := c.getName(data)
+	Data   interface{}
-	if name == "" {
-		name = "_null_"
-	}
-	logger.Logger.Tracef("==> RabbitMQ(%v): %#v", name, data)
-	Send(name, data)
 }
 
-// WriteMQData rabbitMQ消息
+func (c *MessageMgr) Send(data interface{}, name ...string) error {
-func (c *MessageMgr) WriteMQData(name string, data interface{}) {
+	if msg, ok := data.(*RabbitMQData); ok {
-	logger.Logger.Tracef("==> RabbitMQ(%v): %#v", name, data)
+		return Send(msg.MQName, msg.Data)
-	Send(name, data)
+	}
+	if len(name) > 0 && name[0] != "" {
+		return Send(name[0], data)
+	}
+	key := c.getName(data)
+	if key == "" {
+		key = "_null_"
+	}
+	return Send(key, data)
 }
 
 type RegisterHandlerParam struct {
@@ -90,7 +95,7 @@ func (c *MessageMgr) RegisterHandler(param *RegisterHandlerParam) {
 		F: param.Handler,
 	}
 
-	RegisterSubscriber(param.Name, func(e broker.Event) (err error) {
+	internal.RegisterSubscriber(param.Name, func(e broker.Event) (err error) {
 		msg := e.Message()
 		if msg != nil {
 			defer func() {
@@ -126,16 +131,56 @@ func RegisterMessage(param *RegisterMessageParam) {
 
 // Write 发送消息
 // 默认队列名称规则：队列前缀_消息结构体名称
-func Write(data interface{}) {
+func Write(data interface{}, name ...string) error {
-	MessageMgrSingle.Write(data)
+	return MessageMgrSingle.Send(data, name...)
-}
-
-// 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/internal/constant.go

@@ -0,0 +1,20 @@
+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/internal/consumer.go

@@ -1,4 +1,4 @@
-package mq
+package internal
 
 import (
 	"fmt"
@@ -10,8 +10,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/common"
 )
 
 /*
@@ -19,7 +17,6 @@ import (
 */
 
 var (
-	globalConsumer *RabbitMQConsumer
 	subscriberLock sync.RWMutex
 	subscriber     = make(map[string][]*Subscriber)
 )
@@ -56,7 +53,7 @@ func NewRabbitMQConsumer(url string, exchange rabbitmq.Exchange, opts ...broker.
 }
 
 func (c *RabbitMQConsumer) Start() error {
-	if ok, _ := common.PathExists(BackupPath); !ok {
+	if ok, _ := PathExists(BackupPath); !ok {
 		os.MkdirAll(BackupPath, os.ModePerm)
 	}
 
@@ -144,19 +141,3 @@ 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/publisher.go

@@ -1,4 +1,4 @@
-package mq
+package internal
 
 import (
 	"encoding/json"
@@ -12,24 +12,14 @@ 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{}
@@ -121,7 +111,7 @@ func (p *RabbitMQPublisher) publishRoutine() {
 }
 
 func (p *RabbitMQPublisher) Start() (err error) {
-	if ok, _ := common.PathExists(BackupPath); !ok {
+	if ok, _ := PathExists(BackupPath); !ok {
 		err = os.MkdirAll(BackupPath, os.ModePerm)
 		if err != nil {
 			return
@@ -201,26 +191,3 @@ 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
-}

mq/keyconf.go

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

worldsrv/action_bag.go

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

worldsrv/action_phonelottery.go

@@ -3,13 +3,16 @@ 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 {
@@ -195,7 +198,7 @@ func (this *CSPhoneLotteryHandler) Process(s *netlib.Session, packetid int, data
 		if err != nil {
 			return err
 		}
-		LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 1, 0, 0, 0))
+		mq.Write(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 1, 0, 0, 0))
 		for _, v := range items {
 			tp1 := int32(-1)
 			if v.ItemId == common.ItemIDCoin {
@@ -204,8 +207,7 @@ func (this *CSPhoneLotteryHandler) Process(s *netlib.Session, packetid int, data
 				tp1 = model.SystemFreeGive_CoinType_Diamond
 			}
 			if !p.IsRob && tp1 >= 0 {
-				LogChannelSingleton.WriteMQData(
+				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLottery, tp1, v.ItemNum))
-					model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLottery, tp1, v.ItemNum))
 			}
 		}
 	}

worldsrv/action_player.go

@@ -23,6 +23,7 @@ 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"
@@ -1409,7 +1410,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 {
-					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(pi.SnId, pi.Name, pi.Platform, pi.Channel, model.SystemFreeGive_GiveType_NewPlayer, model.SystemFreeGive_CoinType_Coin, int64(model.GameParamData.NewPlayerCoin)))
+					mq.Write(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,
@@ -1427,7 +1428,7 @@ func (this *CSPlayerRegisterHandler) Process(s *netlib.Session, packetid int, da
 					Remark:     "",
 				})
 				if log != nil {
-					LogChannelSingleton.WriteLog(log)
+					mq.Write(log)
 				}
 			}
 			return player_proto.OpResultCode_OPRC_Sucess
@@ -2074,7 +2075,7 @@ func CSPlayerData(s *netlib.Session, packetid int, data interface{}, sid int64)
 			player.addLotteryCount(20)
 			player.InitLotteryStatus = true
 			player.dirty = true
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(player.SnId, player.Platform, "", 3, 20, 5, 0))
+			mq.Write(model.GeneratePhoneLottery(player.SnId, player.Platform, "", 3, 20, 5, 0))
 		}
 
 		// rehold player
@@ -3082,7 +3083,7 @@ func CSUpdateAttribute(s *netlib.Session, packetId int, data interface{}, sid in
 					N:        int64(stepId),
 				}
 				b, _ := json.Marshal(d)
-				LogChannelSingleton.WriteLog(&model.ClientLogMysql{
+				mq.Write(&model.ClientLogMysql{
 					Data:     string(b),
 					Platform: p.Platform,
 					Snid:     p.SnId,

worldsrv/action_rankmatch.go

@@ -8,6 +8,7 @@ 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"
 )
@@ -240,13 +241,14 @@ 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 {
-						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
+						mq.Write(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 {
-						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel,
+						mq.Write(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,6 +12,7 @@ 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"
@@ -115,7 +116,7 @@ func init() {
 			}
 		}
 		//更新玩家等级排行榜
-		LogChannelSingleton.WriteLog(&model.PlayerLevelInfo{
+		mq.Write(&model.PlayerLevelInfo{
 			SnId:     p.SnId,
 			Name:     p.Name,
 			Level:    p.Level,
@@ -134,7 +135,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 {
-						LogChannelSingleton.WriteLog(log)
+						mq.Write(log)
 					}
 				}
 			}

worldsrv/action_task.go

@@ -8,6 +8,7 @@ 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"
 )
@@ -137,8 +138,7 @@ func SendReward(p *Player, m map[int64]int64, tp int32) {
 			tp = model.SystemFreeGive_CoinType_Diamond
 		}
 		if !p.IsRob && tp >= 0 && giveType >= 0 {
-			LogChannelSingleton.WriteMQData(
+			mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp, v.ItemNum))
-				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()
 			}
-			LogChannelSingleton.WriteLog(taskLog)
+			mq.Write(taskLog)
 			ret.OpCode = taskproto.OpResultCode_OPRC_Success
 		}
 	}

worldsrv/action_welfare.go

@@ -12,6 +12,7 @@ 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"
@@ -578,8 +579,7 @@ func SendPermitReward(p *Player, m map[int64]int64, tp int32) {
 			tp1 = model.SystemFreeGive_CoinType_Diamond
 		}
 		if !p.IsRob && tp1 >= 0 && giveType >= 0 {
-			LogChannelSingleton.WriteMQData(
+			mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, tp1, v.ItemNum))
-				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())
 			// 兑换记录
-			LogChannelSingleton.WriteLog(&model.BackendPermitExchange{
+			mq.Write(&model.BackendPermitExchange{
 				Platform: p.Platform,
 				StartTs:  startTs,
 				SnId:     p.SnId,

worldsrv/bagmgr.go

@@ -14,6 +14,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"
 	playerproto "mongo.games.com/game/protocol/player"
@@ -99,7 +100,7 @@ func init() {
 				} else {
 					permitScore = param.ItemNum
 				}
-				LogChannelSingleton.WriteLog(&model.BackendPermitJoin{
+				mq.Write(&model.BackendPermitJoin{
 					Platform: p.Platform,
 					StartTs:  PlatformMgrSingleton.GetConfig(p.Platform).PermitStartTs,
 					SnId:     p.SnId,
@@ -130,7 +131,7 @@ func init() {
 			startTs := PlatformMgrSingleton.GetConfig(p.Platform).PermitStartTs
 			if item != nil && item.ItemNum > 0 && startTs > 0 {
 				// 赛季积分排行榜
-				LogChannelSingleton.WriteLog(&model.PermitScore{
+				mq.Write(&model.PermitScore{
 					Platform: p.Platform,
 					SnId:     p.SnId,
 					Name:     p.Name,
@@ -703,7 +704,7 @@ func (this *BagMgr) AddItems(param *model.AddItemParam) (*BagInfo, bag.OpResultC
 			RoomConfigId: param.RoomConfigId,
 		})
 		if log != nil {
-			LogChannelSingleton.WriteLog(log)
+			mq.Write(log)
 			ts = log.Ts
 			newBagInfo.LogId = log.LogId.Hex()
 		}
@@ -877,7 +878,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 {
-						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Coin, int64(jyb.Award.Coin)))
+						mq.Write(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,
@@ -887,7 +888,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 {
-						LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ActJybAward, model.SystemFreeGive_CoinType_Diamond, int64(jyb.Award.Diamond)))
+						mq.Write(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,
@@ -895,7 +896,7 @@ func (this *BagMgr) VerifyUpJybInfo(p *Player, args *model.VerifyUpJybInfoArgs)
 					})
 				}
 				// 领取日志
-				LogChannelSingleton.WriteLog(jybLog)
+				mq.Write(jybLog)
 				p.dirty = true
 				pack.GainItem.Coin = jyb.Award.Coin
 				pack.GainItem.Diamond = jyb.Award.Diamond

worldsrv/etcd.go

@@ -14,6 +14,7 @@ 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"
@@ -265,7 +266,7 @@ func platformConfigEvent(ctx context.Context, completeKey string, isInit bool, e
 		}
 		f2 := func() {
 			if st.StartTs > 0 {
-				LogChannelSingleton.WriteLog(&model.BackendPermitCycle{
+				mq.Write(&model.BackendPermitCycle{
 					Platform: config.Platform,
 					StartTs:  st.StartTs,
 					EndTs:    st.EndTs - 1,

worldsrv/invitecode.go

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

worldsrv/logchannel.go

@@ -1,75 +0,0 @@
-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,13 +1,8 @@
 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"
@@ -15,19 +10,39 @@ import (
 )
 
 func init() {
-	mq.RegisterSubscriber(mq.WorldLogCoinAck, func(e broker.Event) (err error) {
+	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.LoginLogCollName, Data: &model.LoginLog{}})
-		msg := e.Message()
+	mq.RegisterMessage(&mq.RegisterMessageParam{Name: mq.DBGiveLog, Data: &model.CoinGiveLog{}})
-		if msg != nil {
+	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.CoinLogCollName, Data: &model.CoinLog{}})
-			defer func() {
+	mq.RegisterMessage(&mq.RegisterMessageParam{Name: model.ItemLogCollName, Data: &model.ItemLog{}})
-				e.Ack()
+	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{}})
+}
 
-			var log model.CoinLog
+func init() {
-			err = json.Unmarshal(msg.Body, &log)
+	// 金币对账
-			if err != nil {
+	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 {
 				return
 			}
-
 			//通知主线程执行后续操作
 			core.CoreObject().SendCommand(basic.CommandWrapper(func(o *basic.Object) error {
 				player := PlayerMgrSington.GetPlayerBySnId(log.SnId)
@@ -38,26 +53,19 @@ func init() {
 				return nil
 			}), true)
 			return
-		}
+		},
-		return nil
+	})
-	}, broker.Queue(mq.WorldLogCoinAck), rabbitmq.DurableQueue())
 
 	// 邀请数据同步
-	mq.RegisterSubscriber(mq.WorldInviteAck, func(e broker.Event) error {
+	mq.RegisterHandler(&mq.RegisterHandlerParam{
-		msg := e.Message()
+		Name: mq.WorldInviteAck,
-		if msg != nil {
+		Data: &model.EvtInviteAckMsg{},
-			defer func() {
+		Handler: func(data interface{}) (err error) {
-				e.Ack()
+			log, ok := data.(*model.EvtInviteAckMsg)
-			}()
+			if !ok || log == nil {
-
+				return
-			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)
@@ -84,7 +92,7 @@ func init() {
 				}
 				return nil
 			}), true)
-		}
+			return
-		return nil
+		},
-	}, broker.Queue(mq.WorldInviteAck), rabbitmq.DurableQueue())
+	})
 }

worldsrv/permitmgr.go

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

worldsrv/player.go

@@ -25,6 +25,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"
 	hallproto "mongo.games.com/game/protocol/gamehall"
@@ -346,7 +347,7 @@ func (this *Player) OnLogined() {
 			this.addLotteryCount(20)
 			this.InitLotteryStatus = true
 			this.dirty = true
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 20, 5, 0))
+			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 20, 5, 0))
 		}
 
 		//登录日志
@@ -355,7 +356,7 @@ func (this *Player) OnLogined() {
 		if logState != nil {
 			clog = logState.clog
 		}
-		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeLogin, this.Tel, this.Ip,
+		mq.Write(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))
 
@@ -428,7 +429,7 @@ func (this *Player) OnRehold() {
 		if logState != nil {
 			clog = logState.clog
 		}
-		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeRehold, this.Tel, this.Ip,
+		mq.Write(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))
 
@@ -1117,9 +1118,8 @@ func (this *Player) GetMessageAttach(id string) {
 										tp = model.SystemFreeGive_CoinType_Diamond
 									}
 									if tp != -1 {
-										LogChannelSingleton.WriteMQData(
+										mq.Write(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel,
-											model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel,
+											model.SystemFreeGive_GiveType_TaskPermitRank, tp, int64(msg.Params[i+1])))
-												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 {
-								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
+								mq.Write(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 {
-								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(this.SnId, this.Name, this.Platform, this.Channel, model.SystemFreeGive_GiveType_MailSystemGive,
+								mq.Write(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
 		}
-		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeDrop, this.Tel, this.Ip,
+		mq.Write(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 {
-		LogChannelSingleton.WriteLog(model.NewLoginLog(this.SnId, common.LoginLogTypeLogout, this.Tel, this.Ip,
+		mq.Write(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 {
-			LogChannelSingleton.WriteLog(log)
+			mq.Write(log)
 		}
 	}
 }
@@ -1767,7 +1767,7 @@ func (this *Player) AddCoin(num, add int64, gainWay int32, oper, remark string)
 				Remark:     remark,
 			})
 			if log != nil {
-				LogChannelSingleton.WriteLog(log)
+				mq.Write(log)
 			}
 		}
 	}
@@ -1822,7 +1822,7 @@ func (this *Player) AddCoinAsync(num, add int64, gainWay int32, oper, remark str
 				Remark:     remark,
 			})
 			if log != nil {
-				LogChannelSingleton.WriteLog(log)
+				mq.Write(log)
 			}
 		}
 	}
@@ -2908,7 +2908,7 @@ func (this *Player) ReportLoginEvent() {
 				this.UpgradeTime = this.CreateTime
 			}
 		}
-		LogChannelSingleton.WriteMQData(model.GenerateLogin(model.CreatePlayerLoginEvent(this.SnId,
+		mq.Write(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)))
 		//登录通知
@@ -3255,7 +3255,7 @@ func (this *Player) TryRetrieveLostGameCoin(sceneid int) {
 	logProbe.RestCount = 0 //this.Coin
 	logProbe.SeqNo = this.GameCoinTs
 	logProbe.RoomId = int32(sceneid)
-	LogChannelSingleton.WriteLog(logProbe)
+	mq.Write(logProbe)
 	this.SendDiffData()
 }
 
@@ -3484,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每日礼包")
-			LogChannelSingleton.WriteLog(log)
+			mq.Write(log)
 
 		case 1:
 			var items []*model.Item
@@ -3508,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等级礼包")
-			LogChannelSingleton.WriteLog(log)
+			mq.Write(log)
 		}
 	}
 	send()
@@ -3653,7 +3653,7 @@ func (this *Player) DoShopInfo(info *model.DbShop, isLogin bool) {
 		// 通行证
 		if info.PageId == ShopPagePermit {
 			this.Permit = info.CreateTs.Local()
-			LogChannelSingleton.WriteLog(&model.BackendPermitJoin{
+			mq.Write(&model.BackendPermitJoin{
 				Platform: this.Platform,
 				StartTs:  PlatformMgrSingleton.GetConfig(this.Platform).PermitStartTs,
 				SnId:     this.SnId,
@@ -3767,7 +3767,7 @@ func (this *Player) OnlineLogLogin() {
 	}
 	if this.onlineLog == nil {
 		this.onlineLog = model.NewOnlineLog(bson.NewObjectId(), this.SnId, common.LoginLogTypeLogin, this.Platform)
-		LogChannelSingleton.WriteLog(*this.onlineLog)
+		mq.Write(*this.onlineLog)
 	}
 }
 
@@ -3777,7 +3777,7 @@ func (this *Player) OnlineLogRehold() {
 	}
 	if this.onlineLog == nil {
 		this.onlineLog = model.NewOnlineLog(bson.NewObjectId(), this.SnId, common.LoginLogTypeRehold, this.Platform)
-		LogChannelSingleton.WriteLog(*this.onlineLog)
+		mq.Write(*this.onlineLog)
 	}
 }
 
@@ -3788,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()
-		LogChannelSingleton.WriteLog(*this.onlineLog)
+		mq.Write(*this.onlineLog)
 		this.onlineLog = nil
 	}
 }
@@ -3800,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()
-		LogChannelSingleton.WriteLog(*this.onlineLog)
+		mq.Write(*this.onlineLog)
 		this.onlineLog = nil
 	}
 }
@@ -3879,7 +3879,7 @@ func (this *Player) SendRankSeason() {
 }
 
 func (this *Player) SendPlayerCoin() {
-	LogChannelSingleton.WriteLog(&model.RankPlayerCoin{
+	mq.Write(&model.RankPlayerCoin{
 		Platform: this.Platform,
 		SnId:     this.SnId,
 		Name:     this.Name,
@@ -3969,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)
-			LogChannelSingleton.WriteLog(log)
+			mq.Write(log)
 		}
 		//判断是否是礼包周卡
 		if shopInfo.Page == ShopPageGift {
@@ -4193,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
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 4, 0))
+			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 4, 0))
 		}
 
 	case common.TaskTypeWinOrLose:
@@ -4202,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
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 3, 0))
+			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 3, 0))
 		}
 
 	case common.TaskTypeTienlenCount:
@@ -4211,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
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 2, 0))
+			mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, int(count), 2, 0))
 		}
 	case common.TaskTypeFirstLogin:
 		this.addLotteryCount(5)
-		LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 5, 1, 0))
+		mq.Write(model.GeneratePhoneLottery(this.SnId, this.Platform, "", 3, 5, 1, 0))
 	default:
 		logger.Logger.Errorf("手机抽奖任务未找到对应的任务ID: %d", taskId)
 		return

worldsrv/playeronline.go

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

worldsrv/rankmatch.go

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

worldsrv/shopmgr.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/rand"
-	"mongo.games.com/game/protocol/bag"
 	"slices"
 	"strconv"
 	"time"
@@ -17,7 +16,9 @@ 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"
@@ -595,7 +596,7 @@ func (this *ShopMgr) createPhoneScore(p *Player, shopInfo *model.ShopInfo, costN
 		})
 		jsonData, err := json.Marshal(items)
 		if err == nil {
-			LogChannelSingleton.WriteMQData(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 2, 0, 0, int(costNum)))
+			mq.Write(model.GeneratePhoneLottery(p.SnId, p.Platform, string(jsonData), 2, 0, 0, int(costNum)))
 		}
 		for _, v := range items {
 			tp1 := int32(-1)
@@ -605,8 +606,7 @@ func (this *ShopMgr) createPhoneScore(p *Player, shopInfo *model.ShopInfo, costN
 				tp1 = model.SystemFreeGive_CoinType_Diamond
 			}
 			if !p.IsRob && tp1 >= 0 {
-				LogChannelSingleton.WriteMQData(
+				mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_PhoneLotterySwap, tp1, v.ItemNum))
-					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 {
-				LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Coin, addTotal))
+				mq.Write(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 {
-				LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, model.SystemFreeGive_GiveType_ShopAd, model.SystemFreeGive_CoinType_Diamond, addTotal))
+				mq.Write(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,6 +17,7 @@ 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"
@@ -133,7 +134,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 			if err == nil {
 				// 新号赠送日志
 				if !playerData.IsRob {
-					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
+					mq.Write(model.GenerateSystemFreeGive(
 						playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 						model.SystemFreeGive_GiveType_NewPlayer, model.SystemFreeGive_CoinType_Coin,
 						int64(model.GameParamData.NewPlayerCoin)))
@@ -155,7 +156,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 					Remark:     "",
 				})
 				if log != nil {
-					LogChannelSingleton.WriteLog(log)
+					mq.Write(log)
 				}
 			}
 
@@ -167,7 +168,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 						// 金币
 						if coin > 0 {
 							if !playerData.IsRob {
-								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
+								mq.Write(model.GenerateSystemFreeGive(
 									playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 									model.SystemFreeGive_GiveType_BindTel, model.SystemFreeGive_CoinType_Coin, coin))
 							}
@@ -188,13 +189,13 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 								Remark:     "",
 							})
 							if log != nil {
-								LogChannelSingleton.WriteLog(log)
+								mq.Write(log)
 							}
 						}
 						// 钻石
 						if diamond > 0 {
 							if !playerData.IsRob {
-								LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(
+								mq.Write(model.GenerateSystemFreeGive(
 									playerData.SnId, playerData.Name, playerData.Platform, playerData.Channel,
 									model.SystemFreeGive_GiveType_BindTel, model.SystemFreeGive_CoinType_Diamond, diamond))
 							}
@@ -215,7 +216,7 @@ func (t *TaskLogin) Call(o *basic.Object) interface{} {
 								Remark:     "",
 							})
 							if log != nil {
-								LogChannelSingleton.WriteLog(log)
+								mq.Write(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 {
-//				LogChannelSingleton.WriteLog(log)
+//				mq.Write(log)
 //			}
 //		}
 //		tNode.TransRep.RetFiels = p.Coin

worldsrv/welfmgr.go

@@ -10,6 +10,7 @@ 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"
@@ -109,7 +110,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 {
-				LogChannelSingleton.WriteLog(log)
+				mq.Write(log)
 				logger.Logger.Tracef("NewReliefFundLogEx WriteLog snid: %v Coin:%v", p.SnId, pack.Coin)
 			}
 		}
@@ -556,14 +557,14 @@ func DrawWelfareDate(dates []*webapi_proto.WelfareDate, p *Player, gainWay int32
 			p.AddCoin(coin, add, gainWay, oper, remark)
 			if giveType != -1 {
 				if !p.IsRob {
-					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Coin, int64(coin)))
+					mq.Write(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 {
-					LogChannelSingleton.WriteMQData(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Diamond, int64(v.Grade)))
+					mq.Write(model.GenerateSystemFreeGive(p.SnId, p.Name, p.Platform, p.Channel, giveType, model.SystemFreeGive_CoinType_Diamond, int64(v.Grade)))
 				}
 			}
 		case 3: //道具
@@ -1154,7 +1155,7 @@ func (this *WelfareMgr) RecordWelfareLog(p *Player, num, takenum, discount int64
 
 	if log != nil {
 		//logger.Logger.Trace("RecordItemLog 开始记录 道具操作")
-		LogChannelSingleton.WriteLog(log)
+		mq.Write(log)
 	}
 }