Compare commits
3 Commits
d27c0270f8
...
a0050bd44d
| Author | SHA1 | Date |
|---|---|---|
sk
|
a0050bd44d | |
|
sk
|
7de9e8f939 | |
|
sk
|
6f0906c7c0 |
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -2,56 +2,32 @@ 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"
|
||||
ETCDKEY_GAME_MATCH = "/game/game_match"
|
||||
ETCDKEY_ACT_TURNPLATE = "/game/act_turnplate"
|
||||
ETCDKEY_ACT_7SIGN = "/game/act_7sign"
|
||||
ETCDKEY_ACT_BLINDBOX = "/game/act_blindbox"
|
||||
ETCDKEY_ACT_FIRSTPAY = "/game/act_FirstPay"
|
||||
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/" // 个人水池调控配置
|
||||
ETCDKEY_GAME_CONFIG = "/game/plt/gameconfig/" // 游戏管理/全局配置
|
||||
ETCDKEY_ACT_PHONELOTTERY = "/game/act_phoneLottery"
|
||||
ETCDKEY_ChannelSwitch = "/game/channel/switch" // 渠道开关
|
||||
ETCDKEY_PLATFORM_PREFIX = "/game/plt/config/"
|
||||
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_PROMOTER_PREFIX = "/game/plt/promoter/"
|
||||
ETCDKEY_ACT_GIVE_PREFIX = "/game/plt/actgive/"
|
||||
ETCDKEY_SHOP_EXCHANGE = "/game/exchange_shop"
|
||||
ETCDKEY_GAME_NOTICE = "/game/common_notice"
|
||||
ETCDKEY_SHOP_ITEM = "/game/item_shop"
|
||||
ETCDKEY_GAME_MATCH = "/game/game_match"
|
||||
ETCDKEY_ACT_TURNPLATE = "/game/act_turnplate"
|
||||
ETCDKEY_ACT_7SIGN = "/game/act_7sign"
|
||||
ETCDKEY_ACT_BLINDBOX = "/game/act_blindbox"
|
||||
ETCDKEY_ACT_FIRSTPAY = "/game/act_FirstPay"
|
||||
ETCDKEY_ACT_CONTINUOUSPAY = "/game/act_ContinuousPay"
|
||||
ETCDKEY_ACT_Collect = "/game/act_collect"
|
||||
ETCDKEY_VIP_CFG = "/game/VIPcfg"
|
||||
ETCDKEY_PACKAGE_ENTRYSWITCH = "/game/plt/entryswitch/" //界面入口开关
|
||||
ETCDKEY_CHESSRANK_CFG = "/game/plt/chessrank/" // 象棋段位配置
|
||||
ETCDKEY_PLAYERPOOL = "/game/plt/playerpool/" // 个人水池调控配置
|
||||
ETCDKEY_GAME_CONFIG = "/game/plt/gameconfig/" // 游戏管理/全局配置
|
||||
ETCDKEY_ACT_PHONELOTTERY = "/game/act_phoneLottery"
|
||||
ETCDKEY_ChannelSwitch = "/game/channel/switch" // 渠道开关
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 != nil {
|
||||
return
|
||||
if err = p.Connect(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue