离线加币问题

dbproxy/main.go

@@ -12,6 +12,7 @@ import (
 
 	_ "mongo.games.com/game"
 	"mongo.games.com/game/common"
+	_ "mongo.games.com/game/dbproxy/mq"
 	"mongo.games.com/game/dbproxy/svc"
 	"mongo.games.com/game/model"
 	"mongo.games.com/game/mq"

etcd/keyconf.go

@@ -2,40 +2,17 @@ package etcd
 
 const (
 	// 系统配置
-	ETCDKEY_SYS_ROOT_PREFIX      = "/sys/"
 	ETCDKEY_SYS_PLT_DBCFG_PREFIX = "/sys/plt/dbcfg/"
 
 	//业务配置
-	ETCDKEY_ROOT_PREFIX                = "/game/"
 	ETCDKEY_PLATFORM_PREFIX     = "/game/plt/config/"
-	ETCDKEY_BULLETIN_PREFIX            = "/game/plt/bulletin/"
-	ETCDKEY_AGENTCUSTOMER_PREFIX       = "/game/plt/agent_customer/"
 	ETCDKEY_GAME_CONFIG_GLOBAL  = "/game/plt/game_config_global" // 超管平台游戏开关
 	ETCDKEY_GAMECONFIG_PREFIX   = "/game/plt/game_config/"       // 平台游戏配置
 	ETCDKEY_PACKAGE_PREFIX      = "/game/plt/package/"
 	ETCDKEY_GROUPCONFIG_PREFIX  = "/game/group_config/"
 	ETCDKEY_BLACKLIST_PREFIX    = "/game/plt/black_list/"
-	ETCDKEY_ACT_SIGNIN_PREFIX          = "/game/activity/signin/"
-	ETCDKEY_ACT_TASK_PREFIX            = "/game/activity/task/"
-	ETCDKEY_ACT_GOLDTASK_PREFIX        = "/game/activity/goldtask/"
-	ETCDKEY_ACT_GOLDCOME_PREFIX        = "/game/activity/goldcome/"
-	ETCDKEY_ACT_ONLINEREWARD_PREFIX    = "/game/activity/onlinereward/"
-	ETCDKEY_ACT_LUCKLYTURNTABLE_PREFIX = "/game/activity/lucklyturntable/"
-	ETCDKEY_ACT_YEB_PREFIX             = "/game/activity/yeb/"
-	ETCDKEY_CONFIG_REBATE              = "/game/plt/game_rebate_config/"
 	ETCDKEY_PROMOTER_PREFIX     = "/game/plt/promoter/"
-	ETCDKEY_ACT_VIP_PREFIX             = "/game/plt/actvip/"
-	ETCDKEY_ACT_WEIXIN_SHARE_PREFIX    = "/game/plt/actshare/"
 	ETCDKEY_ACT_GIVE_PREFIX     = "/game/plt/actgive/"
-	ETCDKEY_ACT_PAY_PREFIX             = "/game/plt/payact/"
-	ETCDKEY_ACT_RANDCOIN_PREFIX        = "/game/plt/randcoin/"
-	ETCDKEY_ACT_FPAY_PREFIX            = "/game/plt/fpay/"
-	ETCDKEY_PLATFORM_PROFITCONTROL     = "/game/plt/profitcontrol/"
-	ETCDKEY_MATCH_PROFIX               = "/game/match/"
-	ETCDKEY_ACT_TICKET_PROFIX          = "/game/activity/ticket/"
-	ETCDKEY_ACT_TICKET_RUNNING         = "/game/activity/ticket/running"
-	ETCDKEY_MATCH_GRADESHOP            = "/game/match/gradeshop/"
-	ETCDKEY_CONFIG_LOGICLEVEL          = "/game/logiclevel/"
 	ETCDKEY_SHOP_EXCHANGE       = "/game/exchange_shop"
 	ETCDKEY_GAME_NOTICE         = "/game/common_notice"
 	ETCDKEY_SHOP_ITEM           = "/game/item_shop"
@@ -47,7 +24,6 @@ const (
 	ETCDKEY_ACT_CONTINUOUSPAY   = "/game/act_ContinuousPay"
 	ETCDKEY_ACT_Collect         = "/game/act_collect"
 	ETCDKEY_VIP_CFG             = "/game/VIPcfg"
-	ETCDKEY_WBCtrl_CFG                 = "/game/WBCtrlCfg"
 	ETCDKEY_PACKAGE_ENTRYSWITCH = "/game/plt/entryswitch/" //界面入口开关
 	ETCDKEY_CHESSRANK_CFG       = "/game/plt/chessrank/"   // 象棋段位配置
 	ETCDKEY_PLAYERPOOL          = "/game/plt/playerpool/"  // 个人水池调控配置

mq/consumer.go

@@ -36,7 +36,7 @@ type RabbitMQConsumer struct {
 	exchange rabbitmq.Exchange
 }
 
-func NewRabbitMQConsumer(url string, exchange rabbitmq.Exchange) *RabbitMQConsumer {
+func NewRabbitMQConsumer(url string, exchange rabbitmq.Exchange, opts ...broker.Option) *RabbitMQConsumer {
 	mq := &RabbitMQConsumer{
 		url:      url,
 		exchange: exchange,
@@ -47,6 +47,9 @@ func NewRabbitMQConsumer(url string, exchange rabbitmq.Exchange) *RabbitMQConsum
 	if exchange.Durable {
 		options = append(options, rabbitmq.DurableExchange())
 	}
+	if len(opts) > 0 {
+		options = append(options, opts...)
+	}
 	mq.Broker = rabbitmq.NewBroker(options...)
 	mq.Broker.Init()
 	return mq
@@ -61,6 +64,8 @@ func (c *RabbitMQConsumer) Start() error {
 		return err
 	}
 
+	logger.Logger.Infof("RabbitMQConsumer.Start() url:%s exchange:%s", c.url, c.exchange.Name)
+
 	for topic, ss := range GetSubscribers() {
 		for _, s := range ss {
 			sub, err := c.Subscribe(topic, func(event broker.Event) error {
@@ -68,7 +73,7 @@ func (c *RabbitMQConsumer) Start() error {
 				defer func() {
 					e := recover()
 					if e != nil {
-						logger.Logger.Errorf("RabbitMQConsumer.Subscriber(%s,%v) recover:%v", event.Topic(), event.Message(), e)
+						logger.Logger.Errorf("RabbitMQConsumer.Subscriber() topic:%v message:%v recover:%v", event.Topic(), event.Message(), e)
 					}
 					if err != nil {
 						c.backUp(event, err)
@@ -89,6 +94,7 @@ func (c *RabbitMQConsumer) Start() error {
 }
 
 func (c *RabbitMQConsumer) Stop() error {
+	logger.Logger.Infof("RabbitMQConsumer.Stop() url:%s exchange:%s", c.url, c.exchange.Name)
 	for _, ss := range GetSubscribers() {
 		for _, s := range ss {
 			s.Unsubscribe()
@@ -102,7 +108,7 @@ func (c *RabbitMQConsumer) backUp(e broker.Event, err error) {
 	filePath := fmt.Sprintf(FilePathFormat, BackupPath, e.Topic(), tNow.Format(TimeFormat), tNow.Nanosecond(), rand.Int31n(10000))
 	f, er := os.Create(filePath)
 	if er != nil {
-		logger.Logger.Errorf("RabbitMQPublisher.public(%s,%v) err:%v", e.Topic(), e.Message(), er)
+		logger.Logger.Errorf("RabbitMQPublisher.public() topic:%v message:%v err:%v", e.Topic(), e.Message(), er)
 		return
 	}
 	defer f.Close()
@@ -124,6 +130,7 @@ func RegisterSubscriber(topic string, h broker.Handler, opts ...broker.Subscribe
 	subscriberLock.Lock()
 	subscriber[topic] = append(subscriber[topic], &s)
 	subscriberLock.Unlock()
+	logger.Logger.Infof("RegisterSubscriber topic:%v", topic)
 }
 
 func GetSubscribers() map[string][]*Subscriber {
@@ -139,11 +146,11 @@ func GetSubscribers() map[string][]*Subscriber {
 }
 
 // StartConsumer 启动消费者
-func StartConsumer(url string, exchange string, durableExchange bool) {
+func StartConsumer(url string, exchange string, durableExchange bool, opts ...broker.Option) {
 	StopConsumer()
-	globalConsumer = NewRabbitMQConsumer(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange})
+	globalConsumer = NewRabbitMQConsumer(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, opts...)
 	if err := globalConsumer.Start(); err != nil {
-		panic(fmt.Sprintf("RabbitMQConsumer.Start() err:%v", err))
+		panic(fmt.Sprintf("RabbitMQConsumer.Start() url:%v exchange:%v err:%v", url, exchange, err))
 	}
 }
 

mq/publisher.go

@@ -33,6 +33,7 @@ var globalPublisher *RabbitMQPublisher
 type item struct {
 	topic string
 	msg   interface{}
+	opts  []broker.PublishOption
 }
 
 type RabbitMQPublisher struct {
@@ -44,7 +45,7 @@ type RabbitMQPublisher struct {
 	wg       sync.WaitGroup
 }
 
-func NewRabbitMQPublisher(url string, exchange rabbitmq.Exchange, queueSize int) *RabbitMQPublisher {
+func NewRabbitMQPublisher(url string, exchange rabbitmq.Exchange, queueSize int, opts ...broker.Option) *RabbitMQPublisher {
 	if queueSize <= 0 {
 		queueSize = 1
 	}
@@ -60,17 +61,20 @@ func NewRabbitMQPublisher(url string, exchange rabbitmq.Exchange, queueSize int)
 	if exchange.Durable {
 		options = append(options, rabbitmq.DurableExchange())
 	}
+	if len(opts) > 0 {
+		options = append(options, opts...)
+	}
 	mq.Broker = rabbitmq.NewBroker(options...)
 	mq.Broker.Init()
 	return mq
 }
 
 // 发布消息，同步
-func (p *RabbitMQPublisher) publish(topic string, msg interface{}) (err error) {
+func (p *RabbitMQPublisher) publish(topic string, msg interface{}, opts ...broker.PublishOption) (err error) {
 	defer func() {
 		e := recover()
 		if e != nil {
-			logger.Logger.Errorf("RabbitMQPublisher.public(%s,%v) recover:%v", topic, msg, e)
+			logger.Logger.Errorf("RabbitMQPublisher.public() topic:%v message:%v recover:%v", topic, msg, e)
 		}
 		if err != nil || e != nil {
 			p.backup(topic, msg, err)
@@ -90,9 +94,9 @@ func (p *RabbitMQPublisher) publish(topic string, msg interface{}) (err error) {
 		}
 	}
 
-	err = p.Publish(topic, &broker.Message{Body: buf})
+	err = p.Publish(topic, &broker.Message{Body: buf}, opts...)
 	if err != nil {
-		logger.Logger.Error("RabbitMQPublisher.publish(%s,%v) err:%v", topic, msg, err)
+		logger.Logger.Error("RabbitMQPublisher.publish() topic:%v message:%v err:%v", topic, msg, err)
 		return
 	}
 	return nil
@@ -106,7 +110,7 @@ func (p *RabbitMQPublisher) publishRoutine() {
 		select {
 		case item, ok := <-p.que:
 			if ok {
-				p.publish(item.topic, item.msg)
+				p.publish(item.topic, item.msg, item.opts...)
 			} else {
 				return
 			}
@@ -124,17 +128,19 @@ func (p *RabbitMQPublisher) Start() (err error) {
 		}
 	}
 
-	err = p.Connect()
+	if err = p.Connect(); err != nil {
-	if err != nil {
+		return err
-		return
 	}
 
+	logger.Logger.Infof("RabbitMQPublisher.Start() url:%s exchange:%s", p.url, p.exchange.Name)
+
 	go p.publishRoutine()
 
 	return nil
 }
 
 func (p *RabbitMQPublisher) Stop() error {
+	logger.Logger.Infof("RabbitMQPublisher.Stop() url:%s exchange:%s", p.url, p.exchange.Name)
 	select {
 	case <-p.closed:
 		return ErrClosed
@@ -144,7 +150,7 @@ func (p *RabbitMQPublisher) Stop() error {
 	close(p.closed)
 	close(p.que)
 	for item := range p.que {
-		p.publish(item.topic, item.msg)
+		p.publish(item.topic, item.msg, item.opts...)
 	}
 
 	//等待所有投递出去的任务全部完成
@@ -154,14 +160,14 @@ func (p *RabbitMQPublisher) Stop() error {
 }
 
 // Send 发布消息，异步
-func (p *RabbitMQPublisher) Send(topic string, msg interface{}) (err error) {
+func (p *RabbitMQPublisher) Send(topic string, msg interface{}, opts ...broker.PublishOption) (err error) {
 	select {
 	case <-p.closed:
 		return ErrClosed
 	default:
 	}
 
-	i := &item{topic: topic, msg: msg}
+	i := &item{topic: topic, msg: msg, opts: opts}
 	select {
 	case p.que <- i:
 	default:
@@ -184,7 +190,7 @@ func (p *RabbitMQPublisher) backup(topic string, msg interface{}, err error) {
 	filePath := fmt.Sprintf(FilePathFormat, BackupPath, topic, tNow.Format(TimeFormat), tNow.Nanosecond(), rand.Int31n(10000))
 	f, er := os.Create(filePath)
 	if er != nil {
-		logger.Logger.Errorf("RabbitMQPublisher.public(%s,%v) err:%v", topic, msg, er)
+		logger.Logger.Errorf("RabbitMQPublisher.public() topic:%v message:%v err:%v", topic, msg, er)
 		return
 	}
 	defer f.Close()
@@ -197,11 +203,11 @@ func (p *RabbitMQPublisher) backup(topic string, msg interface{}, err error) {
 }
 
 // StartPublisher 启动发布者
-func StartPublisher(url string, exchange string, durableExchange bool, queueSize int) {
+func StartPublisher(url string, exchange string, durableExchange bool, queueSize int, opts ...broker.Option) {
 	StopPublisher()
-	globalPublisher = NewRabbitMQPublisher(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, queueSize)
+	globalPublisher = NewRabbitMQPublisher(url, rabbitmq.Exchange{Name: exchange, Durable: durableExchange}, queueSize, opts...)
 	if err := globalPublisher.Start(); err != nil {
-		panic(fmt.Sprintf("RabbitMQPublisher.Start() err:%v", err))
+		panic(fmt.Sprintf("RabbitMQPublisher.Start() url:%v exchange:%v err:%v", url, exchange, err))
 	}
 }
 
@@ -212,9 +218,9 @@ func StopPublisher() {
 	}
 }
 
-func Send(topic string, msg interface{}) (err error) {
+func Send(topic string, msg interface{}, opts ...broker.PublishOption) (err error) {
 	if globalPublisher != nil {
-		return globalPublisher.Send(topic, msg)
+		return globalPublisher.Send(topic, msg, opts...)
 	}
 	return ErrClosed
 }

worldsrv/trascate_webapi.go

@@ -1627,17 +1627,21 @@ func init() {
 			CacheDataMgr.CacheBillNumber(billNo, platform) //防止手抖点两下
 			player := PlayerMgrSington.GetPlayerBySnId(member_snid)
 
-			var remainNum = player.Coin
+			var remainNum int64
 			var addcoin, diamond int64 = msg.GetGold(), 0
 			var logtype = int32(common.GainWay_API_AddCoin)
 			if msg.GetLogType() == 1 {
 				addcoin = 0
 				diamond = msg.GetGold()
-				remainNum = player.Diamond
 			}
 			money := msg.Money
 			//玩家在线
 			if player != nil {
+				if msg.GetLogType() == 1 {
+					remainNum = player.Diamond
+				} else {
+					remainNum = player.Coin
+				}
 				//玩家在游戏内
 				if player.scene != nil {
 					CacheDataMgr.ClearCacheBill(billNo, platform)
@@ -1736,6 +1740,12 @@ func init() {
 							}
 						}
 
+						if msg.GetLogType() == 1 {
+							remainNum = findPlayer.Diamond
+						} else {
+							remainNum = findPlayer.Coin
+						}
+
 						//增加帐变记录
 						coinlogex := model.NewCoinLogEx(&model.CoinLogParam{
 							Platform:   findPlayer.Platform,