[MOD]优化

Author: 0990

network/client.go

@@ -11,6 +11,7 @@ type Session interface {
 	SendMsg(msg proto.Message)
 	SendRawMsg(msgID uint32, data []byte)
 	ID() int32
+	Close()
 }
 
 type Client struct {
@@ -84,3 +85,7 @@ func (p *Client) SendRawMsg(msgID uint32, data []byte) {
 func (p *Client) ID() int32 {
 	return p.conn.ID()
 }
+
+func (p *Client) Close() {
+	p.conn.Close()
+}

network/conn.go

@@ -7,4 +7,5 @@ type Conn interface {
 	WriteMsg(args []byte) error
 	LocalAddr() net.Addr
 	ID() int32
+	Close()
 }

rpc/client.go

@@ -150,7 +150,6 @@ func (p *Client) ReadLoop() error {
 		} else if err != nil {
 			return err
 		}
-		util.PrintCurrNano("client data")
 		rpcData := &rpcmsg.Data{}
 		err = proto.Unmarshal(m.Data, rpcData)
 		if err != nil {
@@ -179,7 +178,6 @@ func (p *Client) handle(rpcData *rpcmsg.Data) {
 			return
 		}
 	case rpcmsg.Data_Response:
-		util.PrintCurrNano("client Data_Response")
 		err := p.processor.HandleResponse(seqID, data)
 		if err != nil {
 			logrus.WithError(err)

rpc/rpc.go

@@ -105,3 +105,7 @@ func (p *RPC) GetServerByType(serverType ServerType) Server {
 func (p *RPC) RegisterSend2Session(send2Session func(sesID int32, msgID uint32, data []byte)) {
 	p.client.RegisterSend2Session(send2Session)
 }
+
+func (p *RPC) Session(gateID, sesID int32) Session {
+	return NewSession(p.client, gateID, sesID)
+}

rpc/server.go

@@ -22,6 +22,8 @@ type Server interface {
 
 	//func(*msg.XXX, error))
 	Request(proto.Message, interface{}) error
+
+	ID() int32
 }
 
 type server struct {
@@ -42,6 +44,10 @@ func (p *server) Send(msg proto.Message) {
 	p.rpcClient.SendMsg(p.serverTopic, msg)
 }
 
+func (p *server) ID() int32 {
+	return p.serverid
+}
+
 //func (p *server) Request(msg proto.Message, f func(proto.Message, error)) error {
 //	p.rpcClient.Request(p.serverTopic, msg, f)
 //	return nil
@@ -92,31 +98,45 @@ func (p *requestserver) Answer(msg proto.Message) {
 	p.server.rpcClient.Answer(p.serverTopic, p.seqid, msg)
 }
 
+func (p *requestserver) ID() int32 {
+	return p.serverid
+}
+
 type Session interface {
 	SendMsg(msg proto.Message)
 	SendRawMsg(msgID uint16, data []byte)
+	GateSessionID() GateSessionID
+}
+
+type GateSessionID struct {
+	GateID int32
+	SesID  int32
 }
 
 type session struct {
-	sid       int32
+	gsID      GateSessionID
 	rpcClient *Client
-	gateID    int32
 	gateTopic string
 }
 
 func NewSession(client *Client, gateID int32, sesID int32) Session {
+	g := GateSessionID{GateID: gateID, SesID: sesID}
+
 	return &session{
-		sid:       sesID,
-		gateID:    gateID,
+		gsID:      g,
 		rpcClient: client,
 		gateTopic: fmt.Sprintf("%v", gateID),
 	}
 }
 
 func (p *session) SendMsg(msg proto.Message) {
-	p.rpcClient.RouteGate(p.gateTopic, p.sid, msg)
+	p.rpcClient.RouteGate(p.gateTopic, p.gsID.SesID, msg)
 }
 
 func (p *session) SendRawMsg(msgID uint16, data []byte) {
 
 }
+
+func (p *session) GateSessionID() GateSessionID {
+	return p.gsID
+}

server/gate.go

@@ -66,3 +66,7 @@ func (p *Gate) RouteSessionMsg(msg proto.Message, serverID int32) {
 		p.GetServerById(serverID).RouteSession2Server(s.ID(), msg)
 	})
 }
+
+func (p *Gate) RegisterRawSessionMsgHandler(msg proto.Message, f func(s network.Session, message proto.Message)) {
+	p.networkMgr.RegisterRawSessionMsgHandler(msg, f)
+}

server/server.go

@@ -6,8 +6,9 @@ import (
 )
 
 type Server struct {
-	worker service.Worker
-	rpc    *rpc.RPC
+	worker   service.Worker
+	rpc      *rpc.RPC
+	serverID int32
 }
 
 func NewServer(serverID int32) (*Server, error) {
@@ -18,6 +19,7 @@ func NewServer(serverID int32) (*Server, error) {
 		return nil, err
 	}
 	p.rpc = rpc
+	p.serverID = serverID
 	return p, nil
 }
 
@@ -27,6 +29,10 @@ func (p *Server) Run() {
 	p.rpc.Run()
 }
 
+func (p *Server) Worker() service.Worker {
+	return p.worker
+}
+
 func (p *Server) Post(f func()) {
 	p.worker.Post(f)
 }
@@ -51,3 +57,11 @@ func (p *Server) RegisterSessionMsgHandler(cb interface{}) {
 func (p *Server) RegisterServerHandler(cb interface{}) {
 	p.rpc.RegisterServerMsgHandler(cb)
 }
+
+func (p *Server) ID() int32 {
+	return p.serverID
+}
+
+func (p *Server) RPCSession(s rpc.GateSessionID) rpc.Session {
+	return p.rpc.Session(s.GateID, s.SesID)
+}

service/worker.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"github.com/0990/goserver/util"
 	"runtime/debug"
 	"time"
 )
@@ -10,6 +11,8 @@ type Worker interface {
 	Run()
 	Close()
 	AfterPost(duration time.Duration, f func()) *time.Timer
+	NewTicker(d time.Duration, f func()) *time.Ticker
+	Len() int
 }
 
 //TODO 这里的实现 chan如果塞满会阻塞进程，可对比参照github.com/davyxu/cellnet EventQueue实现，选方案
@@ -19,7 +22,7 @@ type Work struct {
 
 func NewWorker() Worker {
 	p := new(Work)
-	p.funChan = make(chan func(), 1024)
+	p.funChan = make(chan func(), 10240)
 	return p
 }
 
@@ -30,7 +33,7 @@ func (p *Work) Post(f func()) {
 func (p *Work) Run() {
 	go func() {
 		for f := range p.funChan {
-			p.protectedFun(f)
+			util.ProtectedFun(f)
 		}
 	}()
 }
@@ -39,6 +42,9 @@ func (p *Work) Close() {
 	close(p.funChan)
 }
 
+func (p *Work) Len() int {
+	return len(p.funChan)
+}
 func (p *Work) protectedFun(callback func()) {
 	//TODO 每个函数都包装了defer，性能怎样？
 	defer func() {
@@ -49,8 +55,19 @@ func (p *Work) protectedFun(callback func()) {
 	callback()
 }
 
-func (p *Work) AfterPost(duration time.Duration, f func()) *time.Timer {
-	return time.AfterFunc(duration, func() {
+func (p *Work) AfterPost(d time.Duration, f func()) *time.Timer {
+	return time.AfterFunc(d, func() {
 		p.Post(f)
 	})
 }
+
+func (p *Work) NewTicker(d time.Duration, f func()) *time.Ticker {
+	ticker := time.NewTicker(d)
+	go func() {
+		for range ticker.C {
+			p.Post(f)
+			//util.ProtectedFun(f)
+		}
+	}()
+	return ticker
+}

util/fun.go

@@ -0,0 +1,13 @@
+package util
+
+import "runtime/debug"
+
+func ProtectedFun(f func()) {
+	//TODO 每个函数都包装了defer，性能怎样？
+	defer func() {
+		if err := recover(); err != nil {
+			debug.PrintStack()
+		}
+	}()
+	f()
+}