当前位置：首页 > article >正文

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

article 2024/11/17 22:45:05

打开GoLand创建项目

编写Go程序：main.go

package main

import (
	"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 network

import (
	"net"
	"sync"
)

// Server网络服务器
type Server struct {
	callback  ConnCallback  //每一个连接的回调
	protocol  Protocol      //通用协议，用于解析网络协议，处理粘包问题,可以自定义
	exitChan  chan struct{} //退出信号，用于通知服务器退出
	waitGroup *sync.WaitGroup
	closeOnce sync.Once
	listener  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 = listener
	s.waitGroup.Add(1)
	defer func() {
		s.waitGroup.Done()
	}()
	for {
		select {
		case <-s.exitChan:
			return
		default:
		}
		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 network

import (
	"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 network

import (
	"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               *Server
	conn              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) bool
	OnClose(*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 nil
		default:
			return ErrWriteBlocking
		}
	} else {
		select {
		case c.packetSendChan <- p:
			return nil
		case <-time.After(timeout):
			return ErrWriteBlocking
		case <-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:
			return
		case <-c.closeChan:
			return
		default:
		}
		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:
			return
		case <-c.closeChan:
			return
		case 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:
			return
		case <-c.closeChan:
			return
		case p := <-c.packetReceiveChan:
			if c.IsClosed() {
				return
			}
			if !c.callback.OnMessage(c, p) {
				return
			}
		}
	}
}

编写Go程序：gate.go

package gate

import (
	"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 = l
	fmt.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