前言

项目开源地址：https://github.com/llr104/slgserver

比较适合作为go语言入门学习项目或轻量级游戏项目，整体的项目结构和编码质量还是可以的。不过距离商业项目还是差点意思，如服务负载、容灾这块还没有。

项目总共有5个类型独立服务：

gateserver 网关服
loginserver 登录服
chatserver 聊天服
httpserver http服
slgserver 游戏服

其中httpserver就是常规http服务，主要提供一些辅助操作，如账号注册、修改密码什么的。其它4类服务则是用websocket相互连接的。客户端只与网关服相连，网关服负责消息解析、前置处理，然后转发给对应的业务服处理，处理完后响应再原路返回给客户端。

下面我们就已gateserver网关服为例来分析整个项目框架，搞清楚各服务都从哪里启动、消息怎么流转的。如果你不想知道源码实现细节，可直接只看第4步结论。

1. 服务入口

go程序都是从main开始，下面是gateserver的main入口：

// gateserver.go
func main() {
  fmt.Println(os.Getwd())
    
  // 初始化路由(下面第2步) 
  gateserver.Init()
    
  // 启动服务监听(下面第3步)
  needSecret := config.File.MustBool("gateserver", "need_secret", false)  // 读取env.ini配置文件信息
  s := net.NewServer(getGateServerAddr(), needSecret)  // ip/port信息
  s.Router(gateserver.MyRouter)  // 路由信息
  s.SetOnBeforeClose(controller.GHandle.OnServerConnClose) // 连接关闭前回调
  s.Start()  // 开启监听，启动服务
}

接下来我们展开分析下里边具体实现细节。

2. 初始化路由

gateserver.Init()里边具体做了些什么？就只调用了下面这个方法：

func (this*Handle) InitRouter(r *net.Router) {
  this.init()
  g := r.Group("*").Use(middleware.ElapsedTime(), middleware.Log())
  g.AddRouter("*", this.all)
}

先看里面的init，获取到了游戏服、聊天服、登录服连接地址信息。

func (this*Handle) init() {
  this.slgProxy = config.File.MustValue("gateserver", "slg_proxy", "ws://127.0.0.1:8001")
  this.chatProxy = config.File.MustValue("gateserver", "chat_proxy", "ws://127.0.0.1:8002")
  this.loginProxy = config.File.MustValue("gateserver", "login_proxy", "ws://127.0.0.1:8003")
}

接下来这句我们来看看做了什么，g := r.Group("*").Use(middleware.ElapsedTime(), middleware.Log())

首先调用路由对象Router里的Group方法，参数传的*，返回一个Group对象指针，这个Router对象是在gateserver的init.go里创建，通过上面InitRouter方法传递进来的。

func (this*Router) Group(prefix string) *Group{
  g := &Group{prefix: prefix,
    hMap: make(map[string]HandlerFunc),
    hMapMidd: make(map[string][]MiddlewareFunc),
  }

  this.groups = append(this.groups, g)
  return g
}

这个方法创建了一个Group对象，然后把这个对象添加到了Router对象里groups数组里。

type HandlerFunc func(req *WsMsgReq, rsp *WsMsgRsp)
type MiddlewareFunc func(HandlerFunc) HandlerFunc

type Group struct {
  prefix     	string    // 前缀(*)
  hMap       	map[string]HandlerFunc   // 消息存储的map容器
  hMapMidd	map[string][]MiddlewareFunc  // 中间件map容器
  middleware 	[]MiddlewareFunc  // 中间件数组容器
}

然后接着调用Group对象里的Use方法，传递了两个中间件函数middleware.ElapsedTime(), middleware.Log()。把这两个函数添加到了上面Group对象里的middleware数组里。

func (this* Group) Use(middleware ...MiddlewareFunc) *Group{
  this.middleware = append(this.middleware, middleware...)
  return this
}

至此：g := r.Group(“*”).Use(middleware.ElapsedTime(), middleware.Log())干的事情就是，返回了一个Group指针对象，并对这个对象里的prefix字段和middleware字段赋了值。一个赋值字符串*，一个插入了两个中间件函数。

接下来调用g.AddRouter("*", this.all)，首先我们来看传递的参数this.all它是一个方法，参数是消息请求/响应体：

func (this*Handle) all(req *net.WsMsgReq, rsp *net.WsMsgRsp) {
  log.DefaultLog.Info("gateserver handle all begin",
  zap.String("proxyStr", req.Body.Proxy),
  zap.String("msgName", req.Body.Name))
  this.deal(req, rsp)

  if req.Body.Name == "role.enterServer" && rsp.Body.Code == constant.OK  {
    //登录聊天服
    rspObj := &proto.EnterServerRsp{}
    mapstructure.Decode(rsp.Body.Msg, rspObj)
    r := &chat_proto.LoginReq{RId: rspObj.Role.RId, NickName: rspObj.Role.NickName, Token: rspObj.Token}
    reqBody := &net.ReqBody{Seq: 0, Name: "chat.login", Msg: r, Proxy: ""}
    rspBody := &net.RspBody{Seq: 0, Name: "chat.login", Msg: r, Code: 0}
    this.deal(&net.WsMsgReq{Body: reqBody, Conn:req.Conn}, &net.WsMsgRsp{Body: rspBody})
  }

  log.DefaultLog.Info("gateserver handle all end",
  zap.String("proxyStr", req.Body.Proxy),
  zap.String("msgName", req.Body.Name))
}

再来看调用的AddRouter方法：

func (this*Group) AddRouter(name string, handlerFunc HandlerFunc, middleware ...MiddlewareFunc) {
  this.hMap[name] = handlerFunc
  this.hMapMidd[name] = middleware
}

这里对于网关服，没有传递中间件函数，只传递了上面的this.all消息体函数给到了Group对象里的hMap，key=*，value=this.all;

最后，我们再来回答最开始的疑问gateserver.Init()做了什么？我们总结下：

在server/gateserver/controller/handle.go Handle对象里创建了一个Group对象指针，并给这个对象赋了值：

type Group struct {
  prefix     	string    // *
  hMap       	map[string]HandlerFunc   // key:"*" value: this.all
  hMapMidd	  map[string][]MiddlewareFunc  // 空
  middleware 	[]MiddlewareFunc  // middleware.ElapsedTime(), middleware.Log()
}

然后把这个Group对象放到上面初始化传递进来的Router对象里的groups数组里了。

3. 启动服务监听

启动网关服务由下面几句代码完成：

needSecret := config.File.MustBool("gateserver", "need_secret", false)  // 从data/conf/env.ini中读取need_secret字段的值，就是对网络传输数据是否加密标志

// 下面就是创建server对象，并调用它的一些方法启动服务
s := net.NewServer(getGateServerAddr(), needSecret)
s.Router(gateserver.MyRouter)
s.SetOnBeforeClose(controller.GHandle.OnServerConnClose)
s.Start()

s := net.NewServer(getGateServerAddr(), needSecret) 返回server对象，并对里边addr(网关服的ip:port)、needSecret赋值。

type server struct {
  addr        string      // ":8004"
  router      *Router
  needSecret  bool        // true
  beforeClose func (WSConn)
}

func NewServer(addr string, needSecret bool) *server {
  s := server{
    addr: addr,
    needSecret: needSecret,
  }
  return &s
}

s.Router(gateserver.MyRouter)，就是把第二步创建的路由对象赋值给了server对象里的router字段。

1
2
3

func (this*server) Router(router *Router) {
  this.router = router
}

s.SetOnBeforeClose(controller.GHandle.OnServerConnClose)

1
2
3

func (this*server) SetOnBeforeClose(hookFunc func (WSConn))  {
  this.beforeClose = hookFunc
}

把业务控制的server/gateserver/controller/handle.go Handle对象里OnServerConnClose方法赋值给了server对象的beforeClose字段，即用来在网络关闭前做些事情。即释放this.proxy，然后关闭网络。proxy是在上面all方法里的deal里赋值的，this.proxy [ip:port][cid]` = proxyClient(与业务服务连接的客户端对象)

func (this*Handle) OnServerConnClose (conn net.WSConn){
  c, err := conn.GetProperty("cid")
  arr := make([]*net.ProxyClient, 0)

  if err == nil{
    cid := c.(int64)
    this.proxyMutex.Lock()
    for _, m := range this.proxys {
      proxy, ok := m[cid]
      if ok {
        arr = append(arr, proxy)
      }
      delete(m, cid)
    }
    this.proxyMutex.Unlock()
  }

  for _, client := range arr {
    client.Close()
  }
}

至此，创建的server对象的4个数据字段都已经赋了值。

type server struct {
  addr        string        // :8004
  router      *Router       // 第二步的Router
  needSecret  bool          // true
  beforeClose func (WSConn) // 业务控制里的OnServerConnClose
}

最后调用s.Start()启动服务

func (this*server) Start()  {
  log.DefaultLog.Info("server starting")
  // 使用的是go内置的http模块
  http.HandleFunc("/", this.wsHandler)  // 客户端连接请求时，会调用wsHandler
  http.ListenAndServe(this.addr, nil)   // 开启服务监听
}

我们重点看看this.wsHandler，创建websocket对象并初始化事件循环。

func (this*server) wsHandler(resp http.ResponseWriter, req *http.Request) {
  // 创建websocket对象
  wsSocket, err := wsUpgrader.Upgrade(resp, req, nil)  
  if err != nil {
    return
  }

  // websocket用ServerConn对象再包装一层
  conn := ConnMgr.NewConn(wsSocket, this.needSecret)
  log.DefaultLog.Info("client connect", zap.String("addr", wsSocket.RemoteAddr().String()))
  conn.SetRouter(this.router)
  conn.SetOnClose(ConnMgr.RemoveConn)
  conn.SetOnBeforeClose(this.beforeClose)
  conn.Start()
  conn.Handshake()
}

conn.Start()，异步开启两个线程循环，分别处理消息读取和写入。

func (this *ServerConn) Start() {
  go this.wsReadLoop()
  go this.wsWriteLoop()
}

先来看看接收客户端消息：

func (this *ServerConn) wsReadLoop() {
  // defer延迟调用，在wsReadLoop返回前最后调用
  defer func() {
    // recover 是go内置函数，防止宕机，尝试恢复，这个函数只能在defer函数里使用
    if err := recover(); err != nil {
      e := fmt.Sprintf("%v", err)
      log.DefaultLog.Error("wsReadLoop error", zap.String("err", e))
      this.Close()
    }
  }()

  // 消息接收循环
  for {
    // 读一个message
    _, data, err := this.wsSocket.ReadMessage()
    if err != nil {
      break
    }
    
    data, err = util.UnZip(data)
    if err != nil {
      log.DefaultLog.Error("wsReadLoop UnZip error", zap.Error(err))
      continue
    }

    body := &ReqBody{}
    if this.needSecret {
      //检测是否有加密，没有加密发起Handshake,与客户端同步密钥
      if secretKey, err:= this.GetProperty("secretKey"); err == nil {
        key := secretKey.(string)
        d, err := util.AesCBCDecrypt(data, []byte(key), []byte(key), openssl.ZEROS_PADDING)
        if err != nil {
          log.DefaultLog.Error("AesDecrypt error", zap.Error(err))
          this.Handshake()
        }else{
          data = d
        }
      }else{
        log.DefaultLog.Info("secretKey not found client need handshake", zap.Error(err))
        this.Handshake()
        return
      }
    }

    if err := util.Unmarshal(data, body); err == nil {
      req := &WsMsgReq{Conn: this, Body: body}
      rsp := &WsMsgRsp{Body: &RspBody{Name: body.Name, Seq: req.Body.Seq}}
      
      if req.Body.Name == HeartbeatMsg {
        // 心跳信息
        h := &Heartbeat{}
        mapstructure.Decode(body.Msg, h)
        h.STime = time.Now().UnixNano()/1e6
        rsp.Body.Msg = h
      }else{
        if this.router != nil {
          this.router.Run(req, rsp)  // 将解析出来的消息转发出去
        }
      }
      this.outChan <- rsp  // 响应消息放入通道，消息请求到响应是同步的
    }else{
      log.DefaultLog.Error("wsReadLoop Unmarshal error", zap.Error(err))
      this.Handshake()
    }
  }

  this.Close()
}

再来看看发送给客户端的响应消息，这里流程是先收到客户端请求，然后转发到对应的业务服务器，返回后再把回复消息插入通道，最后发给客户端。这里websocket是用的阻塞模式，所以消息都是这么个流程。

func (this *ServerConn) wsWriteLoop() {
  defer func() {
    if err := recover(); err != nil {
      log.DefaultLog.Error("wsWriteLoop error")
      this.Close()
    }
  }()

  for {
    select {
    // 取一个消息
    case msg := <- this.outChan:
      // 写给websocket
      this.write(msg.Body)
    }
  }
}

// 将响应消息加密、打包后发送
func (this *ServerConn) write(msg interface{}) error{
  data, err := util.Marshal(msg)
  if err == nil {
    if this.needSecret {
      if secretKey, err:= this.GetProperty("secretKey"); err == nil {
        key := secretKey.(string)
        data, _ = util.AesCBCEncrypt(data, []byte(key), []byte(key), openssl.ZEROS_PADDING)
      }
    }
  }else {
    log.DefaultLog.Error("wsWriteLoop Marshal body error", zap.Error(err))
    return err
  }

  if data, err := util.Zip(data); err == nil{
    if err := this.wsSocket.WriteMessage(websocket.BinaryMessage, data); err != nil {
      this.Close()
      return err
    }
  }else{
    return err
  }
  return nil
}

总结一下，这一步主要做了创建websocket对象，开启监听端口和消息的读取/发送循环。

4. 客户端与网关及各业务服通信过程

如果你认真跟了前面每一步分析的话，其实整个框架的消息结构已经清晰了。

最后我们来做个总结，总体回顾下：

客户端只与网关服相连，网关服负责消息解析、前置处理，然后转发给对应的业务服处理，处理完后响应再原路返回给客户端。它们之间的连接方式都是websocket。

下面是一条完整请求响应链的流程：

客户端连接网关服，发起消息请求
网关服消息读取循环(wsReadLoop)接收到消息，解析解包后转发消息this.router.Run(req, rsp)

网关消息转发最终实际就是依次调用了middleware.ElapsedTime() , middleware.Log() , gateserver.all()

func (this*Group) applyMiddleware(name string) HandlerFunc {
  h, ok := this.hMap[name]
  if ok == false{
    h, ok = this.hMap["*"]
  }

  if ok {
    // this.middleware存的就是第2步的两个中间件函数，然后h就是来自this.hMap["*"]，即前面的all函数。
    // 从这里可以看出消息在中间件之间流转，其实就是用了go函数的闭包实现的。
    // 函数嵌套，依次调用ElapsedTime、Log、all
    for i := len(this.middleware) - 1; i >= 0; i-- {
      h = this.middleware[i](h)  
    }

    for i := len(this.hMapMidd[name]) - 1; i >= 0; i-- {
      h = this.hMapMidd[name][i](h)
    }
  }

  return h
}
 
h(req, rsp)   // 外部调用返回的h

gateserver.deal()将消息发送到对应的业务服务器(登录服、聊天服、游戏服)，并返回响应数据消息体
最后网关的写消息循环wsWriteLoop将响应消息发给客户端

这就是一个常规的req/resp消息的流程。

最后我们看看服务端怎么主动给客户端推送消息：

// connMgr.go
// 给具体某个玩家推送
func (this *Mgr) PushByRoleId(rid int, msgName string, data interface{}) bool

// 给所有玩家推送
func (this *Mgr) pushAll(msgName string, data interface{})

// 使用方式
net.ConnMgr.PushByRoleId(user.rid, "chat.push", data)

go开源项目slgserver源码分析

前言

1. 服务入口

2. 初始化路由

3. 启动服务监听

4. 客户端与网关及各业务服通信过程