• 8.4 Zinx-V0.8代码实现

    8.4 Zinx-V0.8代码实现

    好了，现在需要将消息队列和多任务worker机制集成到我们Zinx的中了。我们在Server的Start()方法中，在服务端Accept之前，启动Worker工作池。

    zinx/znet/server.go

    1. //开启网络服务
    2. func (s *Server) Start() {
    4.     //...
    6.     //开启一个go去做服务端Linster业务
    7.     go func() {
    8.         //0 启动worker工作池机制
    9.         s.msgHandler.StartWorkerPool()
    11.         //1 获取一个TCP的Addr
    12.         addr, err := net.ResolveTCPAddr(s.IPVersion, fmt.Sprintf("%s:%d", s.IP, s.Port))
    13.         if err != nil {
    14.             fmt.Println("resolve tcp addr err: ", err)
    15.             return
    16.         }
    18.         //...
    19.         //...
    21.         }
    22.     }()
    23. }

    其次，当我们已经得到客户端的连接请求过来数据的时候，我们应该将数据发送给Worker工作池进行处理。

    所以应该在Connection的StartReader()方法中修改：

    zinx/znet/connection.go

    1. /*
    2.     读消息Goroutine，用于从客户端中读取数据
    3.  */
    4. func (c *Connection) StartReader() {
    5.     fmt.Println("Reader Goroutine is  running")
    6.     defer fmt.Println(c.RemoteAddr().String(), " conn reader exit!")
    7.     defer c.Stop()
    9.     for  {
    10.         // 创建拆包解包的对象...
    12.         //读取客户端的Msg head...
    14.         //拆包，得到msgid 和 datalen 放在msg中...
    16.         //根据 dataLen 读取 data，放在msg.Data中...
    18.         //得到当前客户端请求的Request数据
    19.         req := Request{
    20.             conn:c,
    21.             msg:msg,
    22.         }
    24.         if utils.GlobalObject.WorkerPoolSize > 0 {
    25.             //已经启动工作池机制，将消息交给Worker处理
    26.             c.MsgHandler.SendMsgToTaskQueue(&req)
    27.         } else {
    28.             //从绑定好的消息和对应的处理方法中执行对应的Handle方法
    29.             go c.MsgHandler.DoMsgHandler(&req)
    30.         }
    31.     }
    32. }

    这里并没有强制使用多任务Worker机制，而是判断用户配置WorkerPoolSize的个数，如果大于0，那么我就启动多任务机制处理链接请求消息，如果=0或者<0那么，我们依然只是之前的开启一个临时的Goroutine处理客户端请求消息。