上传战斗服务多节点负载均衡

modules/battle/clients.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"go_dreamfactory/lego/core"
+	"go_dreamfactory/lego/sys/log"
 
 	"github.com/gorilla/websocket"
 	"google.golang.org/protobuf/proto"
@@ -18,13 +19,11 @@ import (
 */
 type clientComp struct {
 	modules.MCompGate
-	options      *Options
-	service      core.IService
-	module       *Battle
-	conn         *websocket.Conn
-	seq          uint64
-	pendingmutex sync.Mutex
-	pending      map[uint64]*MessageCall //高并发回调
+	options *Options
+	service core.IService
+	module  *Battle
+	clinets []*client
+	i       int
 }
 
 //组件初始化接口
@@ -32,19 +31,44 @@ func (this *clientComp) Init(service core.IService, module core.IModule, comp co
 	this.MCompGate.Init(service, module, comp, options)
 	this.options = options.(*Options)
 	this.module = module.(*Battle)
-	this.pending = make(map[uint64]*MessageCall)
+	this.clinets = make([]*client, len(this.options.BattleServerAddr))
 	return
 }
 
 func (this *clientComp) Start() (err error) {
 	err = this.MCompGate.Start()
-	dialer := websocket.Dialer{}
-	this.conn, _, err = dialer.Dial(this.options.BattleServerAddr, nil)
+	for i, v := range this.options.BattleServerAddr {
+		if this.clinets[i], err = newClient(v, this.options.Log); err != nil {
+			return
+		}
+	}
 	return
 }
 
-//校验战斗过程
 func (this *clientComp) callBattle(ctx context.Context, method string, req proto.Message, reply proto.Message) (err error) {
+	i := this.i % len(this.clinets)
+	this.i += 1
+	err = this.clinets[i].callBattle(ctx, method, req, reply)
+	return
+}
+
+func newClient(addr string, log log.ILogger) (c *client, err error) {
+	c = &client{log: log, pending: make(map[uint64]*MessageCall)}
+	dialer := websocket.Dialer{}
+	c.conn, _, err = dialer.Dial(addr, nil)
+	return
+}
+
+type client struct {
+	log          log.ILogger
+	conn         *websocket.Conn
+	seq          uint64
+	pendingmutex sync.Mutex
+	pending      map[uint64]*MessageCall //高并发回调
+}
+
+//校验战斗过程
+func (this *client) callBattle(ctx context.Context, method string, req proto.Message, reply proto.Message) (err error) {
 	call := new(MessageCall)
 	call.Done = make(chan *MessageCall, 10)
 	call.Args = req
@@ -74,7 +98,7 @@ func (this *clientComp) callBattle(ctx context.Context, method string, req proto
 		this.pendingmutex.Unlock()
 		if call != nil {
 			call.Error = ctx.Err()
-			call.done(this.options.Log)
+			call.done(this.log)
 		}
 		return ctx.Err()
 	case call := <-call.Done:
@@ -83,7 +107,7 @@ func (this *clientComp) callBattle(ctx context.Context, method string, req proto
 	return
 }
 
-func (this *clientComp) run() {
+func (this *client) run() {
 	var (
 		data []byte
 		msg  *pb.BattleRpcMessage = &pb.BattleRpcMessage{}
@@ -92,23 +116,23 @@ func (this *clientComp) run() {
 locp:
 	for {
 		if _, data, err = this.conn.ReadMessage(); err != nil {
-			this.module.Errorf("client err:%v", err)
+			this.log.Errorf("client err:%v", err)
 			break locp
 		}
 		if err = proto.Unmarshal(data, msg); err != nil {
-			this.module.Errorf("client Unmarshal err:%v", err)
+			this.log.Errorf("client Unmarshal err:%v", err)
 			break locp
 		}
 		go this.handleresponse(msg)
 	}
 }
 
-func (this *clientComp) handleresponse(resp *pb.BattleRpcMessage) {
+func (this *client) handleresponse(resp *pb.BattleRpcMessage) {
 	var call *MessageCall
 	this.pendingmutex.Lock()
 	call = this.pending[resp.Rid]
 	delete(this.pending, resp.Rid)
 	this.pendingmutex.Unlock()
 	call.Error = resp.Data.UnmarshalTo(call.Reply)
-	call.done(this.options.Log)
+	call.done(this.log)
 }

modules/battle/options.go

@@ -13,7 +13,7 @@ type (
 	}
 	Options struct {
 		modules.Options
-		BattleServerAddr string
+		BattleServerAddr []string
 	}
 )
 
@@ -25,7 +25,7 @@ func (this *Options) GetLog() log.ILogger {
 	return this.Log
 }
 
-func (this *Options) GetBattleServerAddr() string {
+func (this *Options) GetBattleServerAddr() []string {
 	return this.BattleServerAddr
 }