golang游戏服务器框架_Go开源游戏服务器框架——Pitaya
简介
Pitaya是一款由国外游戏公司topfreegames使用golang进行编写，易于使用，快速且轻量级的开源分布式游戏服务器框架
Pitaya使用etcd作为默认的服务发现组件，提供使用nats和grpc进行远程调用(server to server)的可选配置，并提供在docker中运行以上组件(etcd、nats)的docker-compose配置
抽象分析
PlayerConn
PlayerConn是一个封装的连接对象，继承net.Conn，并提供一个获取下一个数据包的方法
type PlayerConn interface { GetNextMessage() (b []byte, err error) net.Conn }Acceptor
Acceptor代表一个服务端端口进程，接收客户端连接，并用一个内部Chan来维护这些连接对象
type Acceptor interface { ListenAndServe() Stop() GetAddr() string GetConnChan() chan PlayerConn }Acceptorwrapper
Acceptorwrapper义如其名就是Acceptor的包装器，因为Acceptor的通过Chan来保存连接
所以wrapper可以通过遍历这个Chan来实时包装这些连接
type Wrapper interface { Wrap(acceptor.Acceptor) acceptor.Acceptor }Agent
Agent是一个服务端的应用层连接对象，包含了：
Session信息
服务器预发送消息队列
拆解包对象
最后心跳时间
停止发送心跳的chan
关闭发送数据的chan
全局的关闭信号
连接对象
Agent当前状态
… ..
type ( // Agent corresponds to a user and is used for storing raw Conn information Agent struct { Session *session.Session // session appDieChan chan bool // app die channel chDie chan struct{} // wait for close chSend chan pendingWrite // push message queue chStopHeartbeat chan struct{} // stop heartbeats chStopWrite chan struct{} // stop writing messages closeMutex sync.Mutex conn net.Conn // low-level conn fd decoder codec.PacketDecoder // binary decoder encoder codec.PacketEncoder // binary encoder heartbeatTimeout time.Duration lastAt int64 // last heartbeat unix time stamp messageEncoder message.Encoder ... ... state int32 // current agent state }
pendingWrite struct { ctx context.Context data []byte err error } )Component
Component代表业务组件，提供若干个接口
通过Component生成处理请求的Service
type Component interface { Init() AfterInit() BeforeShutdown() Shutdown() }Handler、Remote、Service
Handler和Remote分别代表本地逻辑执行器和远程逻辑执行器
Service是一组服务对象，包含若干Handler和Remote
这里有个温柔的细节——Receiver reflect.Value
pitaya的设计者为了降低引用，采取在逻辑执行器中保留方法的Receiver以达到在Handler和Remote对象中，只需要保存类型的Method，而无需保存带对象引用的Value.Method
type ( //Handler represents a message.Message's handler's meta information. Handler struct { Receiver reflect.Value // receiver of method Method reflect.Method // method stub Type reflect.Type // low-level type of method IsRawArg bool // whether the data need to serialize MessageType message.Type // handler allowed message type (either request or notify) }
//Remote represents remote's meta information. Remote struct { Receiver reflect.Value // receiver of method Method reflect.Method // method stub HasArgs bool // if remote has no args we won't try to serialize received data into arguments Type reflect.Type // low-level type of method }
// Service implements a specific service, some of it's methods will be // called when the correspond events is occurred. Service struct { Name string // name of service Type reflect.Type // type of the receiver Receiver reflect.Value // receiver of methods for the service Handlers map[string]*Handler // registered methods Remotes map[string]*Remote // registered remote methods Options options // options } )Modules
Modules模块和Component结构一致，唯一的区别在于使用上
Modules主要是面向系统的一些全局存活的对象
方便在统一的时机，集中进行启动和关闭
type Base struct{}
func (c *Base) Init() error { return nil }
func (c *Base) AfterInit() {}
func (c *Base) BeforeShutdown() {}
func (c *Base) Shutdown() error { return nil }集中管理的对象容器在外部module.go中定义
var ( modulesMap = make(map[string]interfaces.Module) modulesArr = []moduleWrapper{} )
type moduleWrapper struct { module interfaces.Module name string }HandleService
HandleService就是服务端的主逻辑对象，负责处理一切数据包
chLocalProcess用于保存待处理的客户端数据包
chRemoteProcess用于保存待处理的来自其他服务器的数据包
services注册了处理客户端的服务
内部聚合一个RemoteService对象，专门负责处理服务器间的数据包
type ( HandlerService struct { appDieChan chan bool // die channel app chLocalProcess chan unhandledMessage // channel of messages that will be processed locally chRemoteProcess chan unhandledMessage // channel of messages that will be processed remotely decoder codec.PacketDecoder // binary decoder encoder codec.PacketEncoder // binary encoder heartbeatTimeout time.Duration messagesBufferSize int remoteService *RemoteService serializer serialize.Serializer // message serializer server *cluster.Server // server obj services map[string]*component.Service // all registered service messageEncoder message.Encoder metricsReporters []metrics.Reporter }
unhandledMessage struct { ctx context.Context agent *agent.Agent route *route.Route msg *message.Message } )RemoteService
RemoteService中维护服务发现和注册提供的远程服务
type RemoteService struct { rpcServer cluster.RPCServer serviceDiscovery cluster.ServiceDiscovery serializer serialize.Serializer encoder codec.PacketEncoder rpcClient cluster.RPCClient services map[string]*component.Service // all registered service router *router.Router messageEncoder message.Encoder server *cluster.Server // server obj remoteBindingListeners []cluster.RemoteBindingListener }Timer
Timer模块中维护一个全局定时任务管理者，使用线程安全的map来保存定时任务，通过time.Ticker的chan信号来定期触发
var ( Manager = &struct { incrementID int64 timers sync.Map ChClosingTimer chan int64 ChCreatedTimer chan *Timer }{}
Precision = time.Second
GlobalTicker *time.Ticker )pipeline
pipeline模块提供全局钩子函数的配置
BeforeHandler 在业务逻辑之前执行
AfterHandler 在业务逻辑之后执行
var ( BeforeHandler = &pipelineChannel{} AfterHandler = &pipelineAfterChannel{} )
type ( HandlerTempl func(ctx context.Context, in interface{}) (out interface{}, err error)AfterHandlerTempl func(ctx context.Context, out interface{}, err error) (interface{}, error)pipelineChannel struct { Handlers []HandlerTempl }
pipelineAfterChannel struct { Handlers []AfterHandlerTempl } )框架流程
app.go是系统启动的入口
创建HandlerService
并根据启动模式如果是集群模式创建RemoteService
开启服务端事件监听
开启监听服务器关闭信号的Chan
var ( app = &App{ ... .. }
remoteService *service.RemoteService handlerService *service.HandlerService ) func Start() { ... .. if app.serverMode == Cluster { ... .. app.router.SetServiceDiscovery(app.serviceDiscovery)
remoteService = service.NewRemoteService( app.rpcClient, app.rpcServer, app.serviceDiscovery, app.router, ... .. )
app.rpcServer.SetPitayaServer(remoteService)
initSysRemotes() }
handlerService = service.NewHandlerService( app.dieChan, app.heartbeat, app.server, remoteService, ... .. )
... ..
listen() ... .. // stop server select { case logger.Log.Warn("the app will shutdown in a few seconds") case s := logger.Log.Warn("got signal: ", s, ", shutting down...") close(app.dieChan) } ... .. }listen方法也就是开启服务，具体包括以下步骤：
1.注册Component
2.注册定时任务的GlobalTicker
3.开启Dispatch处理业务和定时任务(ticket)的goroutine
4.开启acceptor处理连接的goroutine
5.开启主逻辑的goroutine
6.注册Modules
func listen() { startupComponents()
timer.GlobalTicker = time.NewTicker(timer.Precision)
logger.Log.Infof("starting server %s:%s", app.server.Type, app.server.ID) for i := 0; i "pitaya.concurrency.handler.dispatch"); i++ { go handlerService.Dispatch(i) } for _, acc := range app.acceptors { a := acc go func() { for conn := range a.GetConnChan() { go handlerService.Handle(conn) } }()
go func() { a.ListenAndServe() }()
logger.Log.Infof("listening with acceptor %s on addr %s", reflect.TypeOf(a), a.GetAddr()) } ... .. startModules()
logger.Log.Info("all modules started!")
app.running = true }startupComponents对Component进行初始化
然后把Component注册到handlerService和remoteService上
func startupComponents() { // component initialize hooks for _, c := range handlerComp { c.comp.Init() }
// component after initialize hooks for _, c := range handlerComp { c.comp.AfterInit() }
// register all components for _, c := range handlerComp { if err := handlerService.Register(c.comp, c.opts); err != nil { logger.Log.Errorf("Failed to register handler: %s", err.Error()) } }
// register all remote components for _, c := range remoteComp { if remoteService == nil { logger.Log.Warn("registered a remote component but remoteService is not running! skipping...") } else { if err := remoteService.Register(c.comp, c.opts); err != nil { logger.Log.Errorf("Failed to register remote: %s", err.Error()) } } } ... .. }比如HandlerService的注册，反射得到component类型的全部方法，判断isHandlerMethod就加入services里面
并聚合Component对象的反射Value对象为全部Handler的Method Receiver，减少了对象引用
func NewService(comp Component, opts []Option) *Service { s := &Service{ Type: reflect.TypeOf(comp), Receiver: reflect.ValueOf(comp), } ... .. return s }
func (h *HandlerService) Register(comp component.Component, opts []component.Option) error { s := component.NewService(comp, opts) ... .. if err := s.ExtractHandler(); err != nil { return err }
h.services[s.Name] = s for name, handler := range s.Handlers { handlers[fmt.Sprintf("%s.%s", s.Name, name)] = handler } return nil } func (s *Service) ExtractHandler() error { typeName := reflect.Indirect(s.Receiver).Type().Name() ... .. s.Handlers = suitableHandlerMethods(s.Type, s.Options.nameFunc) ... .. for i := range s.Handlers { s.Handlers[i].Receiver = s.Receiver } return nil } func suitableHandlerMethods(typ reflect.Type, nameFunc func(string) string) map[string]*Handler { methods := make(map[string]*Handler) for m := 0; m method := typ.Method(m) mt := method.Type mn := method.Name if isHandlerMethod(method) { ... .. handler := &Handler{ Method: method, IsRawArg: raw, MessageType: msgType, } ... .. methods[mn] = handler } } return methods }handlerService.Dispatch方法负责各种业务的处理，包括：
1.处理chLocalProcess中的本地Message
2.使用remoteService处理chRemoteProcess中的远程Message
3.在定时ticket到达时调用timer.Cron执行定时任务
4.管理定时任务的创建
5.管理定时任务的删除
func (h *HandlerService) Dispatch(thread int) { defer timer.GlobalTicker.Stop()
for { select { case lm := metrics.ReportMessageProcessDelayFromCtx(lm.ctx, h.metricsReporters, "local") h.localProcess(lm.ctx, lm.agent, lm.route, lm.msg)
case rm := metrics.ReportMessageProcessDelayFromCtx(rm.ctx, h.metricsReporters, "remote") h.remoteService.remoteProcess(rm.ctx, nil, rm.agent, rm.route, rm.msg)
case // execute cron task timer.Cron()
case t := // new Timers timer.AddTimer(t)
case id := // closing Timers timer.RemoveTimer(id) } } }接下来看看Acceptor的工作，以下为Tcp实现，就是负责接收连接，流入acceptor的Chan
func (a *TCPAcceptor) ListenAndServe() { if a.hasTLSCertificates() { a.ListenAndServeTLS(a.certFile, a.keyFile) return }
listener, err := net.Listen("tcp", a.addr) if err != nil { logger.Log.Fatalf("Failed to listen: %s", err.Error()) } a.listener = listener a.running = true a.serve() } func (a *TCPAcceptor) serve() { defer a.Stop() for a.running { conn, err := a.listener.Accept() if err != nil { logger.Log.Errorf("Failed to accept TCP connection: %s", err.Error()) continue }
a.connChan Conn: conn, } } }前面讲过对于每个Acceptor开启了一个goroutine去处理连接，也就是下面代码
for conn := range a.GetConnChan() { go handlerService.Handle(conn) }所以流入Chan的连接就会被实时的开启一个goroutine去处理，处理过程就是先创建一个Agent对象
并开启一个goroutine给Agent负责维护连接的心跳
然后开启死循环，读取连接的数据processPacket
func (h *HandlerService) Handle(conn acceptor.PlayerConn) { // create a client agent and startup write goroutine a := agent.NewAgent(conn, h.decoder, h.encoder, h.serializer, h.heartbeatTimeout, h.messagesBufferSize, h.appDieChan, h.messageEncoder, h.metricsReporters)
// startup agent goroutine go a.Handle() ... .. for { msg, err := conn.GetNextMessage()
if err != nil { logger.Log.Errorf("Error reading next available message: %s", err.Error()) return }
packets, err := h.decoder.Decode(msg) if err != nil { logger.Log.Errorf("Failed to decode message: %s", err.Error()) return }
if len(packets) 1 { logger.Log.Warnf("Read no packets, data: %v", msg) continue }
// process all packet for i := range packets { if err := h.processPacket(a, packets[i]); err != nil { logger.Log.Errorf("Failed to process packet: %s", err.Error()) return } } } }这时如果使用了pitaya提供的漏桶算法实现的限流wrap来包装acceptor，则会对客户端发送的消息进行限流限速
这里也是灵活利用for循环遍历chan的特性，所以也是实时地对连接进行包装
func (b *BaseWrapper) ListenAndServe() { go b.pipe() b.Acceptor.ListenAndServe() }
// GetConnChan returns the wrapper conn chan func (b *BaseWrapper) GetConnChan() chan acceptor.PlayerConn { return b.connChan }
func (b *BaseWrapper) pipe() { for conn := range b.Acceptor.GetConnChan() { b.connChan } } type RateLimitingWrapper struct { BaseWrapper }
func NewRateLimitingWrapper(c *config.Config) *RateLimitingWrapper { r := &RateLimitingWrapper{} r.BaseWrapper = NewBaseWrapper(func(conn acceptor.PlayerConn) acceptor.PlayerConn { ... .. return NewRateLimiter(conn, limit, interval, forceDisable) }) return r }
func (r *RateLimitingWrapper) Wrap(a acceptor.Acceptor) acceptor.Acceptor { r.Acceptor = a return r }
func (r *RateLimiter) GetNextMessage() (msg []byte, err error) { if r.forceDisable { return r.PlayerConn.GetNextMessage() }
for { msg, err := r.PlayerConn.GetNextMessage() if err != nil { return nil, err }
now := time.Now() if r.shouldRateLimit(now) { logger.Log.Errorf("Data=%s, Error=%s", msg, constants.ErrRateLimitExceeded) metrics.ReportExceededRateLimiting(pitaya.GetMetricsReporters()) continue }
return msg, err } }processPacket对数据包解包后，执行processMessage
func (h *HandlerService) processPacket(a *agent.Agent, p *packet.Packet) error { switch p.Type { case packet.Handshake: ... .. case packet.HandshakeAck: ... .. case packet.Data: if a.GetStatus() return fmt.Errorf("receive data on socket which is not yet ACK, session will be closed immediately, remote=%s", a.RemoteAddr().String()) } msg, err := message.Decode(p.Data) if err != nil { return err } h.processMessage(a, msg) case packet.Heartbeat: // expected } a.SetLastAt() return nil }processMessage中包装数据包为unHandledMessage
根据消息类型，流入chLocalProcess 或者chRemoteProcess 也就转交给上面提到的负责Dispatch的goroutine去处理了
func (h *HandlerService) processMessage(a *agent.Agent, msg *message.Message) { requestID := uuid.New() ctx := pcontext.AddToPropagateCtx(context.Background(), constants.StartTimeKey, time.Now().UnixNano()) ctx = pcontext.AddToPropagateCtx(ctx, constants.RouteKey, msg.Route) ctx = pcontext.AddToPropagateCtx(ctx, constants.RequestIDKey, requestID.String()) tags := opentracing.Tags{ "local.id": h.server.ID, "span.kind": "server", "msg.type": strings.ToLower(msg.Type.String()), "user.id": a.Session.UID(), "request.id": requestID.String(), } ctx = tracing.StartSpan(ctx, msg.Route, tags) ctx = context.WithValue(ctx, constants.SessionCtxKey, a.Session)
r, err := route.Decode(msg.Route) ... .. message := unhandledMessage{ ctx: ctx, agent: a, route: r, msg: msg, } if r.SvType == h.server.Type { h.chLocalProcess } else { if h.remoteService != nil { h.chRemoteProcess } else { logger.Log.Warnf("request made to another server type but no remoteService running") } } }服务器进程启动的最后一步是对全局模块启动
在外部的module.go文件中，提供了对module的全局注册方法、全部顺序启动方法、全部顺序关闭方法
func RegisterModule(module interfaces.Module, name string) error { ... .. }
func startModules() { for _, modWrapper := range modulesArr { modWrapper.module.Init() } for _, modWrapper := range modulesArr { modWrapper.module.AfterInit() } }
func shutdownModules() { for i := len(modulesArr) - 1; i >= 0; i-- { modulesArr[i].module.BeforeShutdown() }
for i := len(modulesArr) - 1; i >= 0; i-- { mod := modulesArr[i].module mod.Shutdown() } }处理细节
localProcess
接下来看看localprocess对于消息的处理细节(为了直观省略部分异常处理代码)
使用processHandlerMessagef方法对包装出来的ctx对象进行业务操作
最终根据消息的类型 notify / Request 区分是否需要响应，执行不同处理
func (h *HandlerService) localProcess(ctx context.Context, a *agent.Agent, route *route.Route, msg *message.Message) { var mid uint switch msg.Type { case message.Request: mid = msg.ID case message.Notify: mid = 0 }
ret, err := processHandlerMessage(ctx, route, h.serializer, a.Session, msg.Data, msg.Type, false) if msg.Type != message.Notify { ... .. err := a.Session.ResponseMID(ctx, mid, ret) ... .. } else { metrics.ReportTimingFromCtx(ctx, h.metricsReporters, handlerType, nil) tracing.FinishSpan(ctx, err) } }processHandlerMessage
这里面负进行业务逻辑
会先调用executeBeforePipeline(ctx, arg)，执行前置的钩子函数
再通过util.Pcall(h.Method, args)反射调用handler方法
再调用executeAfterPipeline(ctx, resp, err)，执行后置的钩子函数
最后调用serializeReturn(serializer, resp)，对请求结果进行序列化
func processHandlerMessage( ctx context.Context, rt *route.Route, serializer serialize.Serializer, session *session.Session, data []byte, msgTypeIface interface{}, remote bool, ) ([]byte, error) { if ctx == nil { ctx = context.Background() } ctx = context.WithValue(ctx, constants.SessionCtxKey, session) ctx = util.CtxWithDefaultLogger(ctx, rt.String(), session.UID())
h, err := getHandler(rt) ... ..
msgType, err := getMsgType(msgTypeIface) ... ..
logger := ctx.Value(constants.LoggerCtxKey).(logger.Logger) exit, err := h.ValidateMessageType(msgType) ... ..
arg, err := unmarshalHandlerArg(h, serializer, data) ... ..
if arg, err = executeBeforePipeline(ctx, arg); err != nil { return nil, err } ... ..
args := []reflect.Value{h.Receiver, reflect.ValueOf(ctx)} if arg != nil { args = append(args, reflect.ValueOf(arg)) }
resp, err := util.Pcall(h.Method, args) if remote && msgType == message.Notify { resp = []byte("ack") }
resp, err = executeAfterPipeline(ctx, resp, err) ... ..
ret, err := serializeReturn(serializer, resp) ... ..
return ret, nil }executeBeforePipeline
实际就是执行pipeline的BeforeHandler
func executeBeforePipeline(ctx context.Context, data interface{}) (interface{}, error) { var err error res := data if len(pipeline.BeforeHandler.Handlers) > 0 { for _, h := range pipeline.BeforeHandler.Handlers { res, err = h(ctx, res) if err != nil { logger.Log.Debugf("pitaya/handler: broken pipeline: %s", err.Error()) return res, err } } } return res, nil }executeAfterPipeline
实际就是执行pipeline的AfterHandler
func executeAfterPipeline(ctx context.Context, res interface{}, err error) (interface{}, error) { ret := res if len(pipeline.AfterHandler.Handlers) > 0 { for _, h := range pipeline.AfterHandler.Handlers { ret, err = h(ctx, ret, err) } } return ret, err }util.pcall里展示了golang反射的一种高级用法
method.Func.Call，第一个参数是Receiver，也就是调用对象方法的实例
这种设计对比直接保存Value对象的method，反射时直接call，拥有的额外好处就是降低了对象引用，方法不和实例绑定
func Pcall(method reflect.Method, args []reflect.Value) (rets interface{}, err error) { ... .. r := method.Func.Call(args) if len(r) == 2 { if v := r[1].Interface(); v != nil { err = v.(error) } else if !r[0].IsNil() { rets = r[0].Interface() } else { err = constants.ErrReplyShouldBeNotNull } } return }
golang游戏服务器框架_Go开源游戏服务器框架——Pitaya相关推荐
- 开源游戏引擎_Hatchit：开源游戏引擎
开源游戏引擎 通过视频游戏,越来越多的学生正在学习开源世界. 像FreeCiv和Minetest这样的开源游戏邀请年轻的玩家来研究源代码,而像SpigotMC这样的项目则使他们能够编写插件来扩展自己喜 ...
- 开源游戏引擎Godot3.2重大更新,支持更华丽的动态富文本特效
开源游戏引擎Godot3.2 开源游戏引擎Godot3.2近期更新,支持BBCode富文本 开源游戏引擎Godot3.2近期更新,支持BBCode富文本 此PR添加了一种支持实时文本效果的新ItemF ...
- python游戏服务器框架_mqant首页、文档和下载 - Golang/python语言开发的分布式游戏服务器框架 - OSCHINA - 中文开源技术交流社区...
mqant mqant 是一款基于 Golang 语言的简洁,高效,高性能的分布式游戏服务器框架,研发的初衷是要实现一款能支持高并发,高性能,高实时性的游戏服务器框架,也希望 mqant 未来能够做即 ...
- 开源游戏服务器框架汇总
转自:https://gameinstitute.qq.com/community/detail/133153 有哪些开源游戏服务器框架,值得学习呢.基于node.js .java.C#.golang ...
- 总结：那些热门的开源游戏服务器框架，还不看你就out了
##前言 作为一名业内资深的游戏开发人员,经常会遇到实习的新同事在工作中会问到这样的问题: 工作中到底有哪些开源游戏服务器框架,该去值得学习呢? 囊括到node.js .java.C#.golang ...
- Go开源游戏服务器框架——Pitaya
Go开源游戏服务器框架--Pitaya 简介 抽象分析 框架流程 处理细节 简介 Pitaya是一款由国外游戏公司topfreegames使用golang进行编写,易于使用,快速且轻量级的开源分布式游 ...
- 开源游戏服务器框架NoahGameFrame（NF）服务器端环境搭建(二)
一.下载NoahGameFrame 1.进入到开源游戏服务器框架NoahGameFrame在GitHub的官方界面NoahGameFrame 2.复制要Checkout的资源目录URL 3.在任意一个 ...
- 网易开源游戏服务器框架-Pomelo实践（一）
Pomelo是网易开发的一款开源游戏服务器框架,出了做游戏的服务器端,他也可以作为一个高效的网站后台.网址是:http://pomelo.netease.com 其实,他官方的文档是中文的,照理说,我 ...
- 开源游戏服务器框架NoahGameFrame（NF）简介(一)
本文介绍的知识点很多都是来自于官方:NoahGameFrame(NF)官网.点击链接如果没用的话,可以在GitHub上搜索NoahGameFrame. 一.NoahGameFrame是什么? Noah ...
最新文章
- Python实训day13am【Python网络爬虫综合大作业PPT】
- 关系数据库NoSQL数据库
- Python3经典100道练习题003
- php输出mysql的数据结构_php课程 13-43 mysql的数据结构是什么
- html 图片切换渐变效果图,CSS3 用CLIP来做图片切换的渐变效果
- python如何将数据保存到本地json文件
- java手机震动_Windows Phone 7 开发 之使手机震动
- 一不小心，老司机又翻车了
- python all 函数_Python all（）函数
- 异动处理中的发票类型应用(Complaint Processing)
- 每个程序员都可以入手的小项目创意大集合
- python关于luminati国外动态代理的使用
- spring入门配置
- 水库水位库容监测系统方案
- VL600威锋typeC 转HDMI转接单芯片方案，支持DP1.4两LANE实现4K60，
- Windows 查看程序ip地址（面对小白）
- day45--冒泡排序
- 外网访问mysql数据库 花生壳内网映射mysql
- 模具设计与制造类毕业论文文献有哪些？
- mysql8最大连接数设置