9.3 链接的带缓冲的发包方法

我们之前给Connection提供了一个发消息的方法SendMsg()，这个是将数据发送到一个无缓冲的channel中msgChan。但是如果客户端链接比较多的话，如果对方处理不及时，可能会出现短暂的阻塞现象，我们可以做一个提供一定缓冲的发消息方法，做一些非阻塞的发送体验。

zinx/ziface/iconnection.go

  1. //定义连接接口
  2. type IConnection interface {
  3.     //启动连接，让当前连接开始工作
  4.     Start()
  5.     //停止连接，结束当前连接状态M
  6.     Stop()
  7.     //从当前连接获取原始的socket TCPConn
  8.     GetTCPConnection() *net.TCPConn
  9.     //获取当前连接ID
  10.     GetConnID() uint32
  11.     //获取远程客户端地址信息
  12.     RemoteAddr() net.Addr
  13.     //直接将Message数据发送数据给远程的TCP客户端(无缓冲)
  14.     SendMsg(msgId uint32, data []byte) error
  15.     //直接将Message数据发送给远程的TCP客户端(有缓冲)
  16.     SendBuffMsg(msgId uint32, data []byte) error   //添加带缓冲发送消息接口
  17. }

zinx/znet/connection.go

  1. type Connection struct {
  2.     //当前Conn属于哪个Server
  3.     TcpServer    ziface.IServer
  4.     //当前连接的socket TCP套接字
  5.     Conn *net.TCPConn
  6.     //当前连接的ID 也可以称作为SessionID，ID全局唯一
  7.     ConnID uint32
  8.     //当前连接的关闭状态
  9.     isClosed bool
  10.     //消息管理MsgId和对应处理方法的消息管理模块
  11.     MsgHandler ziface.IMsgHandle
  12.     //告知该链接已经退出/停止的channel
  13.     ExitBuffChan chan bool
  14.     //无缓冲管道，用于读、写两个goroutine之间的消息通信
  15.     msgChan        chan []byte
  16.     //有关冲管道，用于读、写两个goroutine之间的消息通信
  17.     msgBuffChan chan []byte                          //定义channel成员
  18. }
  20. //创建连接的方法
  21. func NewConntion(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandler ziface.IMsgHandle) *Connection{
  22.     //初始化Conn属性
  23.     c := &Connection{
  24.         TcpServer:server,
  25.         Conn:     conn,
  26.         ConnID:   connID,
  27.         isClosed: false,
  28.         MsgHandler: msgHandler,
  29.         ExitBuffChan: make(chan bool, 1),
  30.         msgChan:make(chan []byte),
  31.         msgBuffChan:make(chan []byte, utils.GlobalObject.MaxMsgChanLen), //不要忘记初始化
  32.     }
  34.     //将新创建的Conn添加到链接管理中
  35.     c.TcpServer.GetConnMgr().Add(c)
  36.     return c
  37. }

然后将SendBuffMsg()方法实现一下：

  1. func (c *Connection) SendBuffMsg(msgId uint32, data []byte) error {
  2.     if c.isClosed == true {
  3.         return errors.New("Connection closed when send buff msg")
  4.     }
  5.     //将data封包，并且发送
  6.     dp := NewDataPack()
  7.     msg, err := dp.Pack(NewMsgPackage(msgId, data))
  8.     if err != nil {
  9.         fmt.Println("Pack error msg id = ", msgId)
  10.         return  errors.New("Pack error msg ")
  11.     }
  13.     //写回客户端
  14.     c.msgBuffChan <- msg
  16.     return nil
  17. }

我们在Writer中也要有对msgBuffChan的数据监控：

  1. /*
  2.     写消息Goroutine， 用户将数据发送给客户端
  3.  */
  4.  func (c *Connection) StartWriter() {
  5.      fmt.Println("[Writer Goroutine is running]")
  6.      defer fmt.Println(c.RemoteAddr().String(), "[conn Writer exit!]")
  8.      for {
  9.          select {
  10.              case data := <-c.msgChan:
  11.                  //有数据要写给客户端
  12.                  if _, err := c.Conn.Write(data); err != nil {
  13.                      fmt.Println("Send Data error:, ", err, " Conn Writer exit")
  14.                      return
  15.                 }
  16.             //针对有缓冲channel需要些的数据处理
  17.              case data, ok:= <-c.msgBuffChan:
  18.                  if ok {
  19.                     //有数据要写给客户端
  20.                     if _, err := c.Conn.Write(data); err != nil {
  21.                         fmt.Println("Send Buff Data error:, ", err, " Conn Writer exit")
  22.                         return
  23.                     }
  24.                 } else {
  25.                     break
  26.                     fmt.Println("msgBuffChan is Closed")
  27.                 }
  28.              case <-c.ExitBuffChan:
  29.                  return
  30.         }
  31.     }
  32.  }