牌类游戏使用微服务重构笔记(八): 游戏网关服务器

栏目: 后端 · 发布时间: 5年前

内容简介:所谓网关,其实就是维持玩家客户端的连接,将玩家发的游戏请求转发到具体后端服务的服务器,具有以下几个功能点:对于http请求来说,micro框架本身已经实现了api网关,可以参阅之前的博客

所谓网关,其实就是维持玩家客户端的连接,将玩家发的游戏请求转发到具体后端服务的服务器,具有以下几个功能点:

  • 长期运行,必须具有较高的稳定性和性能
  • 对外开放,即客户端需要知道网关的IP和端口,才能连接上来
  • 多协议支持
  • 统一入口,架构中可能存在很多后端服务,如果没有一个统一入口,则客户端需要知道每个后端服务的IP和端口
  • 请求转发,由于统一了入口,所以网关必须能将客户端的请求转发到准确的服务上,需要提供路由
  • 无感更新,由于玩家连接的是网关服务器,只要连接不断;更新后端服务器对玩家来说是无感知的,或者感知很少(根据实现方式不同)
  • 业务无关(对于游戏服务器网关不可避免的可能会有一点业务)

对于http请求来说,micro框架本身已经实现了api网关,可以参阅之前的博客

牌类游戏使用微服务重构笔记(二): micro框架简介:micro toolkit

但是对于游戏服务器,一般都是需要长链接的,需要我们自己实现

连接协议

网关本身应该是支持多协议的,这里就以websocket举例说明我重构过程中的思路,其他协议类似。首先选择提供websocket连接的库 推荐使用 melody ,基于 websocket 库,语法非常简单,数行代码即可实现websocket服务器。我们的游戏需要websocket网关的原因在于客户端不支持HTTP2,不能和grpc服务器直连

package main

import (
	"github.com/micro/go-web"
	"gopkg.in/olahol/melody.v1"
	"log"
	"net/http"
)

func main() {
	// New web service
	service := web.NewService(web.Name("go.micro.api.gateway"))

	// parse command line
	service.Init()

	// new melody
	m := melody.New()

	// Handle websocket connection
	service.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		_ = m.HandleRequest(w, r)
	})
	
	// handle connection with new session
	m.HandleConnect(func(session *melody.Session) {
		
	})
	
	// handle disconnection
	m.HandleDisconnect(func(session *melody.Session) {
		
	})
	
	// handle message
	m.HandleMessage(func(session *melody.Session, bytes []byte) {
		
	})

	// run service
	if err := service.Run(); err != nil {
		log.Fatal("Run: ", err)
	}
}
复制代码

请求转发

网关可以收取或发送数据,并且数据结构比较统一都是 []byte ,这一点是不是很像 grpc stream ,因此就可以使用 protobufoneof 特性来定义请求和响应,可参照上期博客

牌类游戏使用微服务重构笔记(六): protobuf爬坑

定义 gateway.basic.proto ,对网关收/发的消息进行归类

message Message {
    oneof message {
        Req req = 1; // 客户端请求 要求响应
        Rsp rsp = 2; // 服务端响应
        Notify notify = 3; // 客户端推送 不要求响应
        Event event = 4; // 服务端推送
        Stream stream = 5; // 双向流请求
        Ping ping = 6; // ping
        Pong pong = 7;// pong
    }
}
复制代码

对于 reqnotify 都是客户端的无状态请求,对应后端的无状态服务器,这里仅需要实现自己的路由规则即可,比如

message Req {
    string serviceName = 1; // 服务名
    string method = 2; // 方法名
    bytes args = 3; // 参数
    google.protobuf.Timestamp timestamp = 4; // 时间戳
    ...
}
复制代码
req-rsp

思路与 micro toolkit 的api网关类似(rpc 处理器),比较简单,可参照之前的博客。

我们的项目对于此类请求都走http了,并没有通过这个网关, 仅有一些基本的 req ,比如 authReq 处理 session 认证。主要考虑是http简单、无状态、好维护,再加上此类业务对实时性要求也不高。

grpc stream转发

游戏服务器一般都是有状态的、双向的、实时性要求较高, req-rsp 模式并不适合,就需要网关进行转发。每添加一种grpc后端服务器,仅需要在 oneof 中添加一个stream来拓展

message Stream {
   oneof stream {
    room.basic.Message roomMessage = 1; // 房间服务器
    game.basic.Message gameMessage = 2; // 游戏服务器
    mate.basic.Message mateMessage = 3; // 匹配服务器
   }
}
复制代码

并且需要定义一个对应的路由请求,来处理转发到哪一台后端服务器上(实现不同也可以不需要),这里会涉及到一点业务,例如

message JoinRoomStreamReq {
    room.basic.RoomType roomType = 1;
    string roomNo = 2;
}
复制代码

这里根据客户端的路由请求的房间号和房间类型,网关来选择正确的房间服务器(甚至可能链接到旧版本的房间服务器上)

选择正确的服务器后,建立stream 双向流

address := "xxxxxxx" // 选择后的服务器地址
ctx := context.Background() // 顶层context
m := make(map[string]string) // some metadata
streamCtx, cancelFunc := context.WithCancel(ctx) // 复制一个子context

// 建立stream 双向流
stream, err := xxxClient.Stream(metadata.NewContext(streamCtx, m), client.WithAddress(address))

// 存储在session上
session.Set("stream", stream)
session.Set("cancelFunc", cancelFunc)

// 启动一个goroutine 收取stream消息并转发
go func(c context.Context, s pb.xxxxxStream) {
    // 退出时关闭 stream
    defer func() {
	session.Set("stream", nil)
	session.Set("cancelFunc", nil)

	if err := s.Close(); err != nil {
		// do something with close err
	}
    }()

    for {
	select {
    	case <-c.Done():
    		// do something with ctx cancel
    		return
    
    	default:
    		res, err := s.Recv()
    		if err != nil {
    			// do something with recv err
    			return
    		}
    
    		// send to session 这里可以通过oneof包装告知客户端是哪个stream发来的消息
            ...
    	}
    }
}(streamCtx, stream)
复制代码

转发就比较简单了,直接上代码

对于某个stream的请求

message Stream {
   oneof stream {
    room.basic.Message roomMessage = 1; // 房间服务器
    game.basic.Message gameMessage = 2; // 游戏服务器
    mate.basic.Message mateMessage = 3; // 匹配服务器
   }
}
复制代码

添加转发代码

s, exits := session.Get("stream")
if !exits {
    return
}

if stream, ok := s.(pb.xxxxStream); ok {
    err := stream.Send(message)
    if err != nil {
    	log.Println("send err:", err)
    	return
    }
}
复制代码

当需要关闭某个stream时, 只需要调用对应的 cancelFunc 即可

if v, e := session.Get("cancelFunc"); e {
    if c, ok := v.(context.CancelFunc); ok {
    	c()
    }
}
复制代码

使用oneOf的好处

由于接收请求的入口统一,使用 oneof 就可以一路 switch case ,每添加一个 req 或者一种 stream 只需要添加一个case, 代码看起来还是比较简单、清爽的

func HandleMessageBinary(session *melody.Session, bytes []byte) {
    var msg pb.Message

    if err := proto.Unmarshal(bytes, &msg); err != nil {
    	// do something
    	return
    }
    
    defer func() {
	err := recover()
	if err != nil {
	    // do something with panic
	}
    }()
    
    switch x := msg.Message.(type) {
    case *pb.Message_Req:
    	handleReq(session, x.Req)
    
    case *pb.Message_Stream:
    	handleStream(session, x.Stream)
    
    case *pb.Message_Ping:
    	handlePing(session, x.Ping)
    
    default:
    	log.Println("unknown req type")
    }
}

func handleStream(session *melody.Session, message *pb.Stream) {
    switch x := message.Stream.(type) {
    case *pb.Stream_RoomMessage:
        handleRoomStream(session, x.RoomMessage)
    
    case *pb.Stream_GameMessage:
        handleGameStream(session, x.GameMessage)
    
    case *pb.Stream_MateMessage:
        handleMateStream(session, x.MateMessage)
    }
}
复制代码

热更新

对于游戏热更新不停服还是挺重要的,我的思路将会在之后的博客里介绍,可以关注一波 嘿嘿

坑!

  • 这样的网关,看似没什么问题,然而跑上一段时间使用 pprof 观测会发现 goroutine 和内存都在缓慢增长,也就是存在 goroutine leak! ,原因在于 micro源码在包装grpc时,没有对关闭stream完善,只有收到 io.EOF 的错误时才会关闭grpc的conn连接
func (g *grpcStream) Recv(msg interface{}) (err error) {
    defer g.setError(err)
    if err = g.stream.RecvMsg(msg); err != nil {
    	if err == io.EOF {
            // #202 - inconsistent gRPC stream behavior
            // the only way to tell if the stream is done is when we get a EOF on the Recv
            // here we should close the underlying gRPC ClientConn
            closeErr := g.conn.Close()
            if closeErr != nil {
                err = closeErr
            }
    	}
    }
    return
}
复制代码

并且有一个TODO

// Close the gRPC send stream
// #202 - inconsistent gRPC stream behavior
// The underlying gRPC stream should not be closed here since the
// stream should still be able to receive after this function call
// TODO: should the conn be closed in another way?
func (g *grpcStream) Close() error {
    return g.stream.CloseSend()
}
复制代码

解决方法也比较简单,自己fork一份源码改一下关闭stream的时候同时关闭conn(我们的业务是可以的因为在grpc stream客户端和服务端均实现收到err后关闭stream),或者等作者更新用更科学的方式关闭

  • melody的session在 getset 数据时会发生map的读写竞争而panic,可以查看 issue ,解决方法也比较简单

一起学习

本人学习golang、micro、k8s、grpc、protobuf等知识的时间较短,如果有理解错误的地方,欢迎批评指正,可以加我微信一起探讨学习

牌类游戏使用微服务重构笔记(八): 游戏网关服务器

以上所述就是小编给大家介绍的《牌类游戏使用微服务重构笔记(八): 游戏网关服务器》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

Android 源码设计模式解析与实战

Android 源码设计模式解析与实战

何红辉、关爱民 / 人民邮电出版社 / 2015-11 / 79.00元

本书专门介绍Android源代码的设计模式,共26章,主要讲解面向对象的六大原则、主流的设计模式以及MVC和MVP模式。主要内容为:优化代码的首步、开闭原则、里氏替换原则、依赖倒置原则、接口隔离原则、迪米特原则、单例模式、Builder模式、原型模式、工厂方法模式、抽象工厂模式、策略模式、状态模式、责任链模式、解释器模式、命令模式、观察者模式、备忘录模式、迭代器模式、模板方法模式、访问者模式、中介......一起来看看 《Android 源码设计模式解析与实战》 这本书的介绍吧!

RGB转16进制工具
RGB转16进制工具

RGB HEX 互转工具

html转js在线工具
html转js在线工具

html转js在线工具