『网络游戏』GoLand服务器框架【01】

打开GoLand创建项目

编写Go程序：main.go

package mainimport ("fmt""newgame/game/gate""os""os/signal""syscall""time"
)var (SinChan   = make(chan os.Signal, 1)closeChan chan struct{}
)func main() {//信号通道，用于接收系统信号signal.Notify(SinChan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)//逻辑更新定时器，每66毫秒更新一次逻辑logicTicker := time.NewTicker(66 * time.Millisecond)//关闭通道，用于通知程序退出closeChan := make(chan struct{})//退出通道，用于通知程序退出完成exitChan := make(chan struct{})go func() {defer func() {exitChan <- struct{}{}}()//程序初始化地方，可以初始化一些全局变量，或者启动一些协程gs := gate.NewGate()gs.Run()//程序大的循环，处理信号，逻辑更新。。。for {select {//接收到关闭通道，退出程序case <-closeChan:goto QUIT//接受到系统信号，处理信号case sig := <-SinChan:fmt.Println("receive signal:", sig)close(closeChan)case <-logicTicker.C://逻辑更新//fmt.Println("logic update")}}QUIT://处理程序退出逻辑return}()//等待程序退出<-exitChan
}

编写Go程序：server.go

package networkimport ("net""sync"
)// Server网络服务器
type Server struct {callback  ConnCallback  //每一个连接的回调protocol  Protocol      //通用协议，用于解析网络协议，处理粘包问题,可以自定义exitChan  chan struct{} //退出信号，用于通知服务器退出waitGroup *sync.WaitGroupcloseOnce sync.Oncelistener  net.Listener //监听器
}// newServer 创建一个网络服务器
func NewServer(callback ConnCallback, protocol Protocol) *Server {return &Server{callback:  callback,protocol:  protocol,exitChan:  make(chan struct{}),waitGroup: &sync.WaitGroup{},}
}type ConnectionCreator func(conn net.Conn, src *Server) *Connection// start 启动服务器
func (s *Server) Start(listener net.Listener, create ConnectionCreator) {s.listener = listeners.waitGroup.Add(1)defer func() {s.waitGroup.Done()}()for {select {case <-s.exitChan:returndefault:}conn, err := listener.Accept()if err != nil {break}s.waitGroup.Add(1)go func() {create(conn, s).Do()s.waitGroup.Done()}()}
}// Stop停止服务器
func (s *Server) Stop(wait bool) {s.closeOnce.Do(func() {close(s.exitChan)s.listener.Close()})if wait {s.waitGroup.Wait()}
}

编写Go程序：protocol.go

package networkimport ("encoding/binary""errors""io"
)// 网络消息序列化接口
type Packet interface {Serialize() []byte
}// 网络协议读取的接口
type Protocol interface {ReadPacket(conn io.Reader) (Packet, error)
}// 程序默认协议结构
type DefaultPacket struct {buff []byte
}// 实现Packet接口
func (dp *DefaultPacket) Serialize() []byte {return dp.buff
}// 获取消息体的二进制数据
func (db *DefaultPacket) GetBody() []byte {return db.buff[4:]
}// 创建一个默认协议的消息包
func NewDefaultPacket(buff []byte) *DefaultPacket {p := &DefaultPacket{}p.buff = make([]byte, 4+len(buff))binary.BigEndian.PutUint32(p.buff[:4], uint32(len(buff)))//拷贝消息体copy(p.buff[4:], buff)return p
}// 默认协议解析
type DefaultProtocol struct {
}// 实现接口
func (dp *DefaultProtocol) ReadPacket(conn io.Reader) (Packet, error) {var (lengthBytes []byte = make([]byte, 4)length      uint32)//读取消息长度if _, err := io.ReadFull(conn, lengthBytes); err != nil {return nil, err}if length = binary.BigEndian.Uint32(lengthBytes); length > 1024 {return nil, errors.New("the size of packet is too large")}//读取消息体buff := make([]byte, length)if _, err := io.ReadFull(conn, buff); err != nil {return nil, err}return NewDefaultPacket(buff), nil
}

编写Go程序：protocol.go

package networkimport ("errors""net""sync""sync/atomic""time"
)// 定义错误
var (ErrConnClosing   = errors.New("Connection is closing")ErrWriteBlocking = errors.New("write packet is blocking")ErrReadBlocking  = errors.New("read packet is blocking")
)type Connection struct {srv               *Serverconn              net.Conn      //原始链接extraData         interface{}   //额外的数据closeOnce         sync.Once     //关闭链接closeFlag         int32         //关闭标志closeChan         chan struct{} //关闭的信号packetSendChan    chan Packet   //发送消息的通道callback          ConnCallback  //回调的接口packetReceiveChan chan Packet   //接收消息的通道fd                uint64        //链接的唯一标识
}// ConnCallback 每一个网络链接回调的接口
type ConnCallback interface {//OnConnect 当有新的链接进来时的回调，true为接收，false拒绝OnConnect(*Connection) bool//OnMessage当读取到一个完整地游戏逻辑消息时被调用，true继续处理，false关闭OnMessage(*Connection, Packet) boolOnClose(*Connection)
}func NewConnection(conn net.Conn, srv *Server) *Connection {c := &Connection{srv:               srv,callback:          srv.callback,conn:              conn,closeChan:         make(chan struct{}),packetSendChan:    make(chan Packet, 100),packetReceiveChan: make(chan Packet, 100),}if s, ok := conn.(*net.TCPConn); !ok {panic("conn is not")} else {c.fd = uint64(s.RemoteAddr().(*net.TCPAddr).Port)}return c
}// GetFd 获取连接的唯一标识
func (c *Connection) GetFd() uint64 {return c.fd
}// close关闭连接
func (c *Connection) Close() {c.closeOnce.Do(func() {atomic.StoreInt32(&c.closeFlag, 1)close(c.closeChan)close(c.packetSendChan)close(c.packetReceiveChan)c.conn.Close()c.callback.OnClose(c)})
}// 判断链接是否关闭
func (c *Connection) IsClosed() bool {return atomic.LoadInt32(&c.closeFlag) == 1
}// 设置连接的回调
func (c *Connection) SetCallback(callback ConnCallback) {c.callback = callback
}// / AsyncWritePacket 异步写入一个数据包，如果超时则返回错误
func (c *Connection) AsyncWritePacket(p Packet, timeout time.Duration) (err error) {if c.IsClosed() {return ErrConnClosing}defer func() {if e := recover(); e != nil {err = ErrWriteBlocking}}()if timeout == 0 {select {case c.packetSendChan <- p:return nildefault:return ErrWriteBlocking}} else {select {case c.packetSendChan <- p:return nilcase <-time.After(timeout):return ErrWriteBlockingcase <-c.closeChan:return ErrConnClosing}}
}
func (c *Connection) Do() {if !c.callback.OnConnect(c) {return}asyncDo(c.handLoop, c.srv.waitGroup)asyncDo(c.readLoop, c.srv.waitGroup)asyncDo(c.writeLoop, c.srv.waitGroup)
}
func asyncDo(fn func(), wg *sync.WaitGroup) {wg.Add(1)go func() {fn()wg.Done()}()
}
func (c *Connection) readLoop() {defer func() {recover()c.Close()}()for {select {case <-c.srv.exitChan:returncase <-c.closeChan:returndefault:}c.conn.SetReadDeadline(time.Now().Add(time.Second * 180))p, err := c.srv.protocol.ReadPacket(c.conn)if err != nil {return}c.packetReceiveChan <- p}
}// 写数据包到链接
func (c *Connection) writeLoop() {defer func() {recover()c.Close()}()for {select {case <-c.srv.exitChan:returncase <-c.closeChan:returncase p := <-c.packetSendChan:if c.IsClosed() {return}c.conn.SetWriteDeadline(time.Now().Add(time.Second * 180))if _, err := c.conn.Write(p.Serialize()); err != nil {return}}}
}
func (c *Connection) handLoop() {defer func() {recover()c.Close()}()for {select {case <-c.srv.exitChan:returncase <-c.closeChan:returncase p := <-c.packetReceiveChan:if c.IsClosed() {return}if !c.callback.OnMessage(c, p) {return}}}
}

编写Go程序：gate.go

package gateimport ("fmt""net""newgame/game/network"
)// 网关服务
func NewGate() *Gate {return &Gate{}}type Gate struct {listener net.Listener
}// Run启动网关服务器
func (g *Gate) Run() {l, e := net.Listen("tcp", ":10087")if e != nil {panic(e.Error())}g.listener = lfmt.Printf("Linten on %s\n", l.Addr().String())server := network.NewServer(g, &network.DefaultProtocol{})go server.Start(l, func(conn net.Conn, i *network.Server) *network.Connection {return network.NewConnection(conn, server)})
}// Stop停止网关服务器
func (g *Gate) Stop() {err := g.listener.Close()if err != nil {panic(err.Error())}
}// OnConnect 连接建立回调
func (g *Gate) OnConnect(conn *network.Connection) bool {fmt.Printf("new connection:%d\n", conn.GetFd())return true
}// OnClose 连接关闭回调
func (g *Gate) OnClose(conn *network.Connection) {}
func (g *Gate) OnMessage(conn *network.Connection, p network.Packet) bool {return true
}

安装telnet