从0到1开发go-tcp框架【3-读写协程分离、引入消息队列、进入连接管理器、引入连接属性】【基础篇完结】

淩亂°似流年 2024-03-22 23:17 9阅读 0赞

## 从0到1开发go-tcp框架【3-读写协程分离、引入消息队列、进入连接管理器、引入连接属性】 ##

### 1 读写协程分离\[v0.7\] ###

> 1.  添加一个Reader和Writer之间通信的channel
> 2.  添加一个Writer goroutine
> 3.  Reader由之前直接发送给客户端改为发送给通信channel
> 4.  启动Reader和Writer一起工作

#### zinx/znet/connection.go ####

package znet
    
    import (
    	"fmt"
    	"github.com/kataras/iris/v12/x/errors"
    	"io"
    	"net"
    )
    
    type Connection struct {
        
    	Conn       *net.TCPConn
    	ConnID     uint32
    	isClosed   bool
    	msgChannel chan []byte
    	//告知当前的连接已经退出/停止（由Reader告知writer退出）
    	ExitChan   chan bool
    	MsgHandler *MsgHandle
    }
    
    func NewConnection(conn *net.TCPConn, connID uint32, msgHandle *MsgHandle) *Connection {
        
    	c := &Connection{
        
    		Conn:       conn,
    		ConnID:     connID,
    		MsgHandler: msgHandle,
    		isClosed:   false,
    		msgChannel: make(chan []byte),
    		ExitChan:   make(chan bool, 1),
    	}
    	return c
    }
    
    func (c *Connection) StartWriter() {
        
    	fmt.Println("[Writer Goroutine is running]")
    	defer fmt.Println("[conn Writer  goroutine exit!]", c.RemoteAddr().String())
    	//不断的阻塞等待channel的消息，然后将channel中的消息写给客户端
    	for {
        
    		select {
        
    		case data := <-c.msgChannel:
    			//有数据写给客户端
    			if _, err := c.Conn.Write(data); err != nil {
        
    				fmt.Println("Send data error , ", err)
    				return
    			}
    		case <-c.ExitChan:
    			//代表reader已经退出，此时writer也需要退出
    			return
    		}
    	}
    }
    
    func (c *Connection) StartReader() {
        
    	fmt.Println("reader goroutine is running...")
    	defer fmt.Println("[Reader goroutine is exit] connID=", c.ConnID, " remote addr is ", c.RemoteAddr().String())
    	defer c.Stop()
    	//读取数据
    	for {
        
    		//创建一个拆包对象
    		dp := NewDataPack()
    		//读取客户端的msg Head 二进制流 8字节
    		headData := make([]byte, dp.GetHeadLen())
    		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
        
    			fmt.Println("read msg head err ", err)
    			break
    		}
    		//拆包，将读取到的headData封装为msg
    		msg, err := dp.UnPack(headData)
    		if err != nil {
        
    			fmt.Println("unpack msg err ", err)
    			break
    		}
    		//根据dataLen，再次读取Data，放在msg.Data中，
    		var data []byte
    		//如果数据包中有数据，则读取
    		if msg.GetMsgLen() > 0 {
        
    			data = make([]byte, msg.GetMsgLen())
    			//将切片data读满
    			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
        
    				fmt.Println("read msg data err ", err)
    				break
    			}
    		}
    		msg.SetData(data)
    
    		//封装请求，改为router处理
    		r := Request{
        
    			conn: c,
    			msg:  msg,
    		}
    		go c.MsgHandler.DoMsgHandler(&r)
    	}
    }
    
    //启动连接
    func (c *Connection) Start() {
        
    	fmt.Printf("ConnID %d is Start...", c.ConnID)
    	//开启读、写
    	go c.StartReader()
    	go c.StartWriter()
    }
    
    //停止连接
    func (c *Connection) Stop() {
        
    	fmt.Println("Connection Stop()...ConnectionID = ", c.ConnID)
    	if c.isClosed {
        
    		return
    	}
    	c.isClosed = true
    	c.Conn.Close()
    	c.ExitChan <- true
    	close(c.msgChannel)
    	close(c.ExitChan)
    }
    
    //获取当前连接的Conn对象
    func (c *Connection) GetTCPConnection() *net.TCPConn {
        
    	return c.Conn
    }
    
    //获取当前连接模块的id
    func (c *Connection) GetConnectionID() uint32 {
        
    	return c.ConnID
    }
    
    //获取远程客户端的TCP状态 IP：Port
    func (c *Connection) RemoteAddr() net.Addr {
        
    	return c.Conn.RemoteAddr()
    }
    
    //发送数据
    func (c *Connection) SendMsg(msgId uint32, data []byte) error {
        
    	if c.isClosed {
        
    		return errors.New("connection closed\n")
    	}
    	//将data进行封包
    	dp := NewDataPack()
    	binaryMsg, err := dp.Pack(NewMessage(msgId, data))
    	if err != nil {
        
    		fmt.Println("Pack error msg id=", msgId)
    		return errors.New("pack error msg")
    	}
    	//将数据发送给客户端
    	if _, err := c.Conn.Write(binaryMsg); err != nil {
        
    		fmt.Println("write msg id ", msgId, " error ", err)
    		return errors.New("conn write err ")
    	}
    	return nil
    }

#### 测试 ####

##### myDemo/ZinxV0.7/client.go #####

*  client0.go

package main
    
    import (
    	"fmt"
    	"io"
    	"myTest/zinx/znet"
    	"net"
    	"time"
    )
    
    /*
    模拟客户端
    */
    func main() {
        
    	fmt.Println("client start...")
    	time.Sleep(time.Second * 1)
    	//1 创建服务器连接
    	conn, err := net.Dial("tcp", "127.0.0.1:8092")
    	if err != nil {
        
    		fmt.Println("client start err ", err)
    		return
    	}
    	for {
        
    		//发送封装后的数据包
    		dp := znet.NewDataPack()
    		binaryMsg, err := dp.Pack(znet.NewMessage(0, []byte("Zinx client0 test msg")))
    		if err != nil {
        
    			fmt.Println("client pack msg err ", err)
    			return
    		}
    		if _, err := conn.Write(binaryMsg); err != nil {
        
    			fmt.Println("client write err ", err)
    			return
    		}
    		//服务器应该给我们回复一个message数据，msgId为1，内容为ping...ping...
    
    		//1 先读取流中的head部分，得到Id和dataLen
    		binaryHead := make([]byte, dp.GetHeadLen())
    		if _, err := io.ReadFull(conn, binaryHead); err != nil {
        
    			fmt.Println("client read head err ", err)
    			break
    		}
    		//将二进制的head拆包到msg中
    		msgHead, err := dp.UnPack(binaryHead)
    		if err != nil {
        
    			fmt.Println("client unpack msgHead err ", err)
    			break
    		}
    		if msgHead.GetMsgLen() > 0 {
        
    			//2 有数据, 再根据dataLen进行二次读取，将data读出来
    			msg := msgHead.(*znet.Message)
    			msg.Data = make([]byte, msg.GetMsgLen())
    			if _, err := io.ReadFull(conn, msg.Data); err != nil {
        
    				fmt.Println("read msg data error ", err)
    				return
    			}
    			fmt.Println("--------> Receive Server msg , ID=", msg.Id, " ,len=", msg.DataLen, " ,data=", string(msg.Data))
    		}
    
    		//cpu阻塞，让出cpu时间片，避免无限for循环导致其他程序无法获取cpu时间片
    		time.Sleep(time.Second * 1)
    	}
    }

*  client1.go

![在这里插入图片描述][0ad78ed681b04114b375519aa7875724.png]

##### myDemo/ZinxV0.7/server.go #####

结果：  
![在这里插入图片描述][964159f564d94508864ac0d9329bb324.png]

*  接受多个客户端也可以  
    ![在这里插入图片描述][45caa17774a0445ea4f5636bbc7041bc.png]
 *  当client0退出时，不会影响client1  
    ![在这里插入图片描述][16d52470510e40beb25b75c5d2974695.png]

### 2 创建消息队列及多任务\[v0.8\] ###

> 1.  创建一个消息队列,MsgHandler消息管理模块增加：TaskQueue、WorkerPoolSize
> 2.  创还能多任务worker的工作池并且启动
> 3.  将之前发送的消息，全部改为把消息发送给消息队列和worker工作池来处理

![在这里插入图片描述][9582aefc318f4cb489dab7813a5b6f64.png]

#### 实现消息队列机制和工作池机制（集成到自定义框架） ####

> 1.  创建一个消息队列：MsgHandler消息管理模块
> 2.  创建多任务worker的工作池并启动
> 3.  将之前发送的消息，全部改为把消息发送给消息队列和worker工作池来处理
> 4.  将消息队列机制集成到Zinx框架中
> 
>  *  开启并调用消息队列及worker工作池
>  *  将从客户端处理的消息，发送给当前Worker的工作池来处理

##### zinx/znet/server.go #####

package znet
    
    import (
    	"fmt"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"net"
    )
    
    type Server struct {
        
    	Name       string
    	IPVersion  string
    	IP         string
    	Port       int
    	MsgHandler *MsgHandle
    }
    
    func NewServer(name string) *Server {
        
    	s := &Server{
        
    		Name:       name,
    		IPVersion:  "tcp4",
    		IP:         util.GlobalObject.Host,
    		Port:       util.GlobalObject.TcpPort,
    		MsgHandler: NewMsgHandle(),
    	}
    	return s
    }
    
    func (s *Server) Start() {
        
    	//启动服务监听端口
    	fmt.Printf("[Zinx] Server Name :%s , listen IP :%v , Port: %d is starting \n", s.Name, s.IP, s.Port)
    	fmt.Printf("[Zinx] Version :%s , MaxConn:%v , MaxPackageSize: %d \n", util.GlobalObject.Version, util.GlobalObject.MaxConn, util.GlobalObject.MaxPackageSize)
    	var cid uint32 = 0
    	go func() {
        
    		//0 开启消息队列及Worker工作池
    		s.MsgHandler.StartWorkerPool()
    
    		addr, err := net.ResolveTCPAddr(s.IPVersion, fmt.Sprintf("%s:%d", s.IP, s.Port))
    		if err != nil {
        
    			fmt.Printf("resolve tcp addr error %v\n", err)
    			return
    		}
    		listener, err := net.ListenTCP(s.IPVersion, addr)
    		if err != nil {
        
    			fmt.Println("listen ", s.IPVersion, " err ", err)
    			return
    		}
    		fmt.Println("[start] Zinx server success ", s.Name, "Listening...")
    		//阻塞连接，处理业务
    		for {
        
    			conn, err := listener.AcceptTCP()
    			if err != nil {
        
    				fmt.Println("Accept err ", err)
    				continue
    			}
    			dealConn := NewConnection(conn, cid, s.MsgHandler)
    			cid++
    			//开启goroutine处理启动当前conn
    			go dealConn.Start()
    		}
    	}()
    }
    
    func (s *Server) Stop() {
        
    
    }
    
    func (s *Server) Serve() {
        
    	s.Start()
    	//阻塞，一直读取客户端所发送过来的消息
    	select {
        }
    }
    
    func (s *Server) AddRouter(msgId uint32, router ziface.IRouter) {
        
    	s.MsgHandler.AddRouter(msgId, router)
    }

##### zinx/znet/connection.go #####

package znet
    
    import (
    	"fmt"
    	"github.com/kataras/iris/v12/x/errors"
    	"io"
    	"myTest/zinx/util"
    	"net"
    )
    
    type Connection struct {
        
    	Conn       *net.TCPConn
    	ConnID     uint32
    	isClosed   bool
    	msgChannel chan []byte
    	//告知当前的连接已经退出/停止（由Reader告知writer退出）
    	ExitChan   chan bool
    	MsgHandler *MsgHandle
    }
    
    func NewConnection(conn *net.TCPConn, connID uint32, msgHandle *MsgHandle) *Connection {
        
    	c := &Connection{
        
    		Conn:       conn,
    		ConnID:     connID,
    		MsgHandler: msgHandle,
    		isClosed:   false,
    		msgChannel: make(chan []byte),
    		ExitChan:   make(chan bool, 1),
    	}
    	return c
    }
    
    func (c *Connection) StartWriter() {
        
    	fmt.Println("[Writer Goroutine is running]")
    	defer fmt.Println("[conn Writer  goroutine exit!]", c.RemoteAddr().String())
    	//不断的阻塞等待channel的消息，然后将channel中的消息写给客户端
    	for {
        
    		select {
        
    		case data := <-c.msgChannel:
    			//有数据写给客户端
    			if _, err := c.Conn.Write(data); err != nil {
        
    				fmt.Println("Send data error , ", err)
    				return
    			}
    		case <-c.ExitChan:
    			//代表reader已经退出，此时writer也需要退出
    			return
    		}
    	}
    }
    
    func (c *Connection) StartReader() {
        
    	fmt.Println("reader goroutine is running...")
    	defer fmt.Println("[Reader goroutine is exit] connID=", c.ConnID, " remote addr is ", c.RemoteAddr().String())
    	defer c.Stop()
    	//读取数据
    	for {
        
    		//创建一个拆包对象
    		dp := NewDataPack()
    		//读取客户端的msg Head 二进制流 8字节
    		headData := make([]byte, dp.GetHeadLen())
    		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
        
    			fmt.Println("read msg head err ", err)
    			break
    		}
    		//拆包，将读取到的headData封装为msg
    		msg, err := dp.UnPack(headData)
    		if err != nil {
        
    			fmt.Println("unpack msg err ", err)
    			break
    		}
    		//根据dataLen，再次读取Data，放在msg.Data中，
    		var data []byte
    		//如果数据包中有数据，则读取
    		if msg.GetMsgLen() > 0 {
        
    			data = make([]byte, msg.GetMsgLen())
    			//将切片data读满
    			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
        
    				fmt.Println("read msg data err ", err)
    				break
    			}
    		}
    		msg.SetData(data)
    
    		//封装请求，改为router处理
    		r := Request{
        
    			conn: c,
    			msg:  msg,
    		}
    		//判断是否开启workerPool,如果没有开启则直接创建协程处理；如果开启则通过workerPool处理
    		if util.GlobalObject.WorkerPoolSize > 0 {
        
    			c.MsgHandler.SendMsgToTaskQueue(&r)
    		} else {
        
    			go c.MsgHandler.DoMsgHandler(&r)
    		}
    	}
    }
    
    //启动连接
    func (c *Connection) Start() {
        
    	fmt.Printf("ConnID %d is Start...", c.ConnID)
    	//开启读、写
    	go c.StartReader()
    	go c.StartWriter()
    }
    
    //停止连接
    func (c *Connection) Stop() {
        
    	fmt.Println("Connection Stop()...ConnectionID = ", c.ConnID)
    	if c.isClosed {
        
    		return
    	}
    	c.isClosed = true
    	c.Conn.Close()
    	c.ExitChan <- true
    	close(c.msgChannel)
    	close(c.ExitChan)
    }
    
    //获取当前连接的Conn对象
    func (c *Connection) GetTCPConnection() *net.TCPConn {
        
    	return c.Conn
    }
    
    //获取当前连接模块的id
    func (c *Connection) GetConnectionID() uint32 {
        
    	return c.ConnID
    }
    
    //获取远程客户端的TCP状态 IP：Port
    func (c *Connection) RemoteAddr() net.Addr {
        
    	return c.Conn.RemoteAddr()
    }
    
    //发送数据
    func (c *Connection) SendMsg(msgId uint32, data []byte) error {
        
    	if c.isClosed {
        
    		return errors.New("connection closed\n")
    	}
    	//将data进行封包
    	dp := NewDataPack()
    	binaryMsg, err := dp.Pack(NewMessage(msgId, data))
    	if err != nil {
        
    		fmt.Println("Pack error msg id=", msgId)
    		return errors.New("pack error msg")
    	}
    	//将数据发送给客户端
    	if _, err := c.Conn.Write(binaryMsg); err != nil {
        
    		fmt.Println("write msg id ", msgId, " error ", err)
    		return errors.New("conn write err ")
    	}
    	return nil
    }

##### zinx/znet/msgHandler.go #####

package znet
    
    import (
    	"fmt"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"strconv"
    )
    
    type MsgHandle struct {
        
    	//msgId与对应的router对应
    	Api map[uint32]ziface.IRouter
    	//负责worker取任务的消息队列
    	TaskQueue []chan ziface.IRequest
    	//业务工作worker池的goroutine数量
    	WorkerPoolSize uint32
    }
    
    func NewMsgHandle() *MsgHandle {
        
    	return &MsgHandle{
        
    		Api:            make(map[uint32]ziface.IRouter),
    		TaskQueue:      make([]chan ziface.IRequest, util.GlobalObject.WorkerPoolSize),
    		WorkerPoolSize: util.GlobalObject.WorkerPoolSize,
    	}
    }
    
    func (mh *MsgHandle) DoMsgHandler(request ziface.IRequest) {
        
    	//判断是否有对应的router
    	if _, ok := mh.Api[request.GetMsgID()]; !ok {
        
    		fmt.Println("msgId ", request.GetMsgID(), "does not exist handler, need to add router")
    		return
    	}
    	//call handler
    	router := mh.Api[request.GetMsgID()]
    	router.PreHandle(request)
    	router.Handler(request)
    	router.PostHandler(request)
    }
    
    func (mh *MsgHandle) AddRouter(msgId uint32, router ziface.IRouter) {
        
    	if _, ok := mh.Api[msgId]; ok {
        
    		//如果已经存在了对应的router，则提示
    		panic("repeat api, msgId = " + strconv.Itoa(int(msgId)))
    	}
    	mh.Api[msgId] = router
    	fmt.Println("msgId ", msgId, "Add router success ")
    }
    
    //启动一个worker工作池（开启工作池的动作只能发生一次，一个zinx框架只能有一个worker工作池）
    func (mh *MsgHandle) StartWorkerPool() {
        
    	for i := 0; i < int(mh.WorkerPoolSize); i++ {
        
    		//开辟任务队列
    		mh.TaskQueue[i] = make(chan ziface.IRequest, util.GlobalObject.MaxWorkerTaskLen)
    		//启动worker
    		go mh.startOneWorker(i, mh.TaskQueue[i])
    	}
    }
    
    func (mh *MsgHandle) startOneWorker(workerId int, taskQueue chan ziface.IRequest) {
        
    	fmt.Println("Worker ID=", workerId, " is started...")
    	for {
        
    		select {
        
    		//从任务队列中取消息（如果有消息过来，出列的就是request，然后执行该request所绑定的业务）
    		case request := <-taskQueue:
    			mh.DoMsgHandler(request)
    		}
    	}
    }
    
    //将消息交给taskQueue，由Worker进行处理
    func (mh *MsgHandle) SendMsgToTaskQueue(request ziface.IRequest) {
        
    	//通过取余数的方式来达到负载均衡
    	workID := request.GetConnection().GetConnectionID() % util.GlobalObject.WorkerPoolSize
    	fmt.Println("Add ConnID=", request.GetConnection().GetConnectionID(),
    		" requestID=", request.GetMsgID(),
    		" workID=", workID)
    	//将消息发送给对应worker的任务队列
    	mh.TaskQueue[workID] <- request
    }

##### zinx/ziface/imsgHandler.go #####

package ziface
    
    type IMsgHandler interface {
        
    	DoMsgHandler(request IRequest)
    	AddRouter(msgId uint32, router IRouter)
    	StartWorkerPool()
    	SendMsgToTaskQueue(request IRequest)
    }

#### 测试 ####

##### myDemo/ZinxV0.8/Server.go #####

> 同myDemo/ZinxV0.7/Server.go，修改一下NewServer时候所传的Zinx的名称即可

##### myDemo/ZinxV0.8/Client.go #####

> 同myDemo/ZinxV0.7/Client.go

##### myDemo/ZinxV0.8/zinx.json #####

{
        
      "Name": "Zinx Server Application",
      "Version": "V0.8",
      "Host": "0.0.0.0",
      "TcpPort": 8092,
      "MaxConn": 30,
      "MaxPackageSize": 1024,
      "WorkerPoolSize": 10
    }

![在这里插入图片描述][14098d4841c04c8aa488185a274d1f8a.png]

![在这里插入图片描述][14c33b84249142219365dd1f3f65d3a3.png]

### 3 连接管理器（connManager）\[v0.9\] ###

#### 3.1 连接管理器(conn)的定义与实现 ####

> 创建一个连接管理模块ConnManager
> 
>  *  添加连接
>  *  删除连接
>  *  根据连接ID查找对应的连接
>  *  总连接个数
>  *  清理全部的连接

#### 3.2 将连接管理模块集成到Zinx框架中 ####

> 1.  给server添加一个ConnMgr属性
> 2.  修改NewServer方法，加入ConnMgr初始化
> 3.  判断当前连接数是否超出最大值MaxConn
> 4.  当server停止的时候（调用server.Stop方法），应该加入ConnMgr.ClearConn()

#### 3.3 提供创建连接/销毁连之前所需的Hook函数 ####

> 给我们自定义框架Zinx提供创建连接之后/销毁连接之前所要处理的一些业务。提供给用户能够注册的Hook函数
> 
>  *  添加OnConnStart()
>  *  添加OnConnStop()

##### zinx/ziface/iserver.go #####

package ziface
    
    type IServer interface {
        
    	Start()
    	Stop()
    	Serve()
    	AddRouter(msgId uint32, router IRouter)
    	GetConnMgr() IConnManager
    	//注册创OnConnStart钩子函数
    	SetOnConnStart(func(conn IConnection))
    	SetOnConnStop(func(conn IConnection))
    	//调用OnConnStart钩子函数
    	CallOnConnStart(conn IConnection)
    	CallOnConnStop(conn IConnection)
    }

##### zinx/ziface/iconnmanager.go #####

package ziface
    
    type IConnManager interface {
        
    	Add(conn IConnection)
    	Remove(conn IConnection)
    	Get(connID uint32) (IConnection, error)
    	Len() int
    	ClearConn()
    }

##### zinx/znet/connmanager.go #####

package znet
    
    import (
    	"fmt"
    	"github.com/kataras/iris/v12/x/errors"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"sync"
    )
    
    type ConnManager struct {
        
    	connections map[uint32]ziface.IConnection //管理的连接集合
    	connLock    sync.RWMutex                  //保护连接集合的读写锁
    }
    
    func NewConnManager() *ConnManager {
        
    	return &ConnManager{
        
    		connections: make(map[uint32]ziface.IConnection, util.GlobalObject.MaxConn),
    	}
    }
    func (cm *ConnManager) Add(conn ziface.IConnection) {
        
    	//添加写锁
    	cm.connLock.Lock()
    	defer cm.connLock.Unlock()
    	cm.connections[conn.GetConnectionID()] = conn
    	fmt.Println("connectionID=", conn.GetConnectionID(), " add to ConnManager success, conn num=", cm.Len())
    }
    
    func (cm *ConnManager) Remove(conn ziface.IConnection) {
        
    	//保护共享资源map
    	cm.connLock.Lock()
    	defer cm.connLock.Unlock()
    	delete(cm.connections, conn.GetConnectionID())
    	fmt.Println("connectionID=", conn.GetConnectionID(), " remote from ConnManager success, conn num=", cm.Len())
    }
    
    func (cm *ConnManager) Get(connID uint32) (ziface.IConnection, error) {
        
    	cm.connLock.RLock()
    	defer cm.connLock.RUnlock()
    	if conn, ok := cm.connections[connID]; ok {
        
    		return conn, nil
    	} else {
        
    		return nil, errors.New("connection NOT FOUND")
    	}
    }
    
    func (cm *ConnManager) Len() int {
        
    	return len(cm.connections)
    }
    
    func (cm *ConnManager) ClearConn() {
        
    	cm.connLock.Lock()
    	defer cm.connLock.Unlock()
    	for connID, conn := range cm.connections {
        
    		//停止连接
    		conn.Stop()
    		//删除连接
    		delete(cm.connections, connID)
    	}
    	fmt.Println("Clear All connections success! conn num=", cm.Len())
    }

##### zinx/znet/connection.go #####

package znet
    
    import (
    	"fmt"
    	"github.com/kataras/iris/v12/x/errors"
    	"io"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"net"
    )
    
    type Connection struct {
        
    	Conn       *net.TCPConn
    	ConnID     uint32
    	isClosed   bool
    	msgChannel chan []byte
    	//告知当前的连接已经退出/停止（由Reader告知writer退出）
    	ExitChan   chan bool
    	MsgHandler *MsgHandle
    	TcpServer  ziface.IServer
    }
    
    func NewConnection(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandle *MsgHandle) *Connection {
        
    	c := &Connection{
        
    		Conn:       conn,
    		ConnID:     connID,
    		MsgHandler: msgHandle,
    		isClosed:   false,
    		msgChannel: make(chan []byte),
    		ExitChan:   make(chan bool, 1),
    		TcpServer:  server,
    	}
    	//将conn添加到connMgr中
    	c.TcpServer.GetConnMgr().Add(c)
    	return c
    }
    
    func (c *Connection) StartWriter() {
        
    	fmt.Println("[Writer Goroutine is running]")
    	defer fmt.Println("[conn Writer  goroutine exit!]", c.RemoteAddr().String())
    	//不断的阻塞等待channel的消息，然后将channel中的消息写给客户端
    	for {
        
    		select {
        
    		case data := <-c.msgChannel:
    			//有数据写给客户端
    			if _, err := c.Conn.Write(data); err != nil {
        
    				fmt.Println("Send data error , ", err)
    				return
    			}
    		case <-c.ExitChan:
    			//代表reader已经退出，此时writer也需要退出
    			return
    		}
    	}
    }
    
    func (c *Connection) StartReader() {
        
    	fmt.Println("reader goroutine is running...")
    	defer fmt.Println("[Reader goroutine is exit] connID=", c.ConnID, " remote addr is ", c.RemoteAddr().String())
    	defer c.Stop()
    	//读取数据
    	for {
        
    		//创建一个拆包对象
    		dp := NewDataPack()
    		//读取客户端的msg Head 二进制流 8字节
    		headData := make([]byte, dp.GetHeadLen())
    		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
        
    			fmt.Println("read msg head err ", err)
    			break
    		}
    		//拆包，将读取到的headData封装为msg
    		msg, err := dp.UnPack(headData)
    		if err != nil {
        
    			fmt.Println("unpack msg err ", err)
    			break
    		}
    		//根据dataLen，再次读取Data，放在msg.Data中，
    		var data []byte
    		//如果数据包中有数据，则读取
    		if msg.GetMsgLen() > 0 {
        
    			data = make([]byte, msg.GetMsgLen())
    			//将切片data读满
    			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
        
    				fmt.Println("read msg data err ", err)
    				break
    			}
    		}
    		msg.SetData(data)
    
    		//封装请求，改为router处理
    		r := Request{
        
    			conn: c,
    			msg:  msg,
    		}
    		//判断是否开启workerPool,如果没有开启则直接创建协程处理；如果开启则通过workerPool处理
    		if util.GlobalObject.WorkerPoolSize > 0 {
        
    			c.MsgHandler.SendMsgToTaskQueue(&r)
    		} else {
        
    			go c.MsgHandler.DoMsgHandler(&r)
    		}
    	}
    }
    
    //启动连接
    func (c *Connection) Start() {
        
    	fmt.Printf("ConnID %d is Start...", c.ConnID)
    	//开启读、写
    	go c.StartReader()
    	go c.StartWriter()
    	//执行钩子函数
    	c.TcpServer.CallOnConnStart(c)
    }
    
    //停止连接
    func (c *Connection) Stop() {
        
    	fmt.Println("Connection Stop()...ConnectionID = ", c.ConnID)
    	if c.isClosed {
        
    		return
    	}
    	c.isClosed = true
    	//连接关闭之前执行hook关闭的钩子函数
    	c.TcpServer.CallOnConnStop(c)
    	c.Conn.Close()
    	c.ExitChan <- true
    
    	//连接conn关闭时，需要从连接管理模块中移除
    	c.TcpServer.GetConnMgr().Remove(c)
    	close(c.msgChannel)
    	close(c.ExitChan)
    }
    
    //获取当前连接的Conn对象
    func (c *Connection) GetTCPConnection() *net.TCPConn {
        
    	return c.Conn
    }
    
    //获取当前连接模块的id
    func (c *Connection) GetConnectionID() uint32 {
        
    	return c.ConnID
    }
    
    //获取远程客户端的TCP状态 IP：Port
    func (c *Connection) RemoteAddr() net.Addr {
        
    	return c.Conn.RemoteAddr()
    }
    
    //发送数据
    func (c *Connection) SendMsg(msgId uint32, data []byte) error {
        
    	if c.isClosed {
        
    		return errors.New("connection closed\n")
    	}
    	//将data进行封包
    	dp := NewDataPack()
    	binaryMsg, err := dp.Pack(NewMessage(msgId, data))
    	if err != nil {
        
    		fmt.Println("Pack error msg id=", msgId)
    		return errors.New("pack error msg")
    	}
    	//将数据发送给客户端
    	if _, err := c.Conn.Write(binaryMsg); err != nil {
        
    		fmt.Println("write msg id ", msgId, " error ", err)
    		return errors.New("conn write err ")
    	}
    	return nil
    }

##### zinx/znet/server.go #####

package znet
    
    import (
    	"fmt"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"net"
    )
    
    type Server struct {
        
    	Name       string
    	IPVersion  string
    	IP         string
    	Port       int
    	MsgHandler *MsgHandle
    	ConnMgr    *ConnManager
    	//创建连接之前的Hook函数
    	OnConnStart func(conn ziface.IConnection)
    	OnConnStop  func(conn ziface.IConnection)
    }
    
    func NewServer(name string) *Server {
        
    	s := &Server{
        
    		Name:       name,
    		IPVersion:  "tcp4",
    		IP:         util.GlobalObject.Host,
    		Port:       util.GlobalObject.TcpPort,
    		MsgHandler: NewMsgHandle(),
    		ConnMgr:    NewConnManager(),
    	}
    	return s
    }
    
    func (s *Server) Start() {
        
    	//启动服务监听端口
    	fmt.Printf("[Zinx] Server Name :%s , listen IP :%v , Port: %d is starting \n", s.Name, s.IP, s.Port)
    	fmt.Printf("[Zinx] Version :%s , MaxConn:%v , MaxPackageSize: %d \n", util.GlobalObject.Version, util.GlobalObject.MaxConn, util.GlobalObject.MaxPackageSize)
    	var cid uint32 = 0
    	go func() {
        
    		//0 开启消息队列及Worker工作池
    		s.MsgHandler.StartWorkerPool()
    
    		addr, err := net.ResolveTCPAddr(s.IPVersion, fmt.Sprintf("%s:%d", s.IP, s.Port))
    		if err != nil {
        
    			fmt.Printf("resolve tcp addr error %v\n", err)
    			return
    		}
    		listener, err := net.ListenTCP(s.IPVersion, addr)
    		if err != nil {
        
    			fmt.Println("listen ", s.IPVersion, " err ", err)
    			return
    		}
    		fmt.Println("[start] Zinx server success ", s.Name, "Listening...")
    		//阻塞连接，处理业务
    		for {
        
    			conn, err := listener.AcceptTCP()
    			if err != nil {
        
    				fmt.Println("Accept err ", err)
    				continue
    			}
    			//判断当前连接数是否超过最大连接数，如果超过则关闭新创建的连接
    			if s.ConnMgr.Len() >= util.GlobalObject.MaxConn {
        
    				//TODO 给客户端返回一个超出最大连接的错误包
    				fmt.Println("-----------------》 Tcp Conn exceed, conn num=", util.GlobalObject.MaxConn)
    				conn.Close()
    				//关闭当前连接，等待下一次连接【如果当前连接数小于最大连接数】
    				continue
    			}
    			dealConn := NewConnection(s, conn, cid, s.MsgHandler)
    			cid++
    			//开启goroutine处理启动当前conn
    			go dealConn.Start()
    		}
    	}()
    }
    
    func (s *Server) Stop() {
        
    	//释放相关资源
    	fmt.Println("[STOP] Zinx server name ", s.Name)
    	s.ConnMgr.ClearConn()
    }
    
    func (s *Server) Serve() {
        
    	s.Start()
    	//阻塞，一直读取客户端所发送过来的消息
    	select {
        }
    }
    
    func (s *Server) GetConnMgr() ziface.IConnManager {
        
    	return s.ConnMgr
    }
    
    func (s *Server) AddRouter(msgId uint32, router ziface.IRouter) {
        
    	s.MsgHandler.AddRouter(msgId, router)
    }
    
    //注册创OnConnStart钩子函数
    func (s *Server) SetOnConnStart(hookFunc func(conn ziface.IConnection)) {
        
    	s.OnConnStart = hookFunc
    }
    
    func (s *Server) SetOnConnStop(hookFunc func(conn ziface.IConnection)) {
        
    	s.OnConnStop = hookFunc
    }
    
    //调用OnConnStart钩子函数
    func (s *Server) CallOnConnStart(conn ziface.IConnection) {
        
    	if s.OnConnStart != nil {
        
    		fmt.Println("---------> call OnConnStart()")
    		s.OnConnStart(conn)
    	}
    }
    
    func (s *Server) CallOnConnStop(conn ziface.IConnection) {
        
    	if s.OnConnStop != nil {
        
    		fmt.Println("----------> call OnConnStop()")
    		s.OnConnStop(conn)
    	}
    }

##### 测试 #####

###### myDemo/ZinxV0.9/Server.go ######

package main
    
    import (
    	"fmt"
    	"myTest/zinx/ziface"
    	"myTest/zinx/znet"
    )
    
    //自定义一个Router，测试路由功能
    type PingRouter struct {
        
    	znet.BaseRouter
    }
    
    func (pr *PingRouter) Handler(request ziface.IRequest) {
        
    	fmt.Println("call router handler...")
    	//先读取客户端数据，再回写ping...ping...ping...
    	fmt.Println("receive from client msgId=", request.GetMsgID(),
    		"data=", string(request.GetData()))
    
    	//回写ping
    	err := request.GetConnection().SendMsg(0, []byte("ping...ping...ping..."))
    	if err != nil {
        
    		fmt.Println(err)
    	}
    }
    
    //定义第二个Router
    type HelloRouter struct {
        
    	znet.BaseRouter
    }
    
    func (hr *HelloRouter) Handler(request ziface.IRequest) {
        
    	fmt.Println("receive from client msgId=", request.GetMsgID(),
    		"data=", string(request.GetData()))
    	err := request.GetConnection().SendMsg(1, []byte("hello zinx, I'm the other handler"))
    	if err != nil {
        
    		fmt.Println(err)
    	}
    }
    
    //连接创建成功之后需要执行的逻辑
    func DoConnBegin(conn ziface.IConnection) {
        
    	fmt.Println("=====>Do Conn Begin...")
    	if err := conn.SendMsg(202, []byte("do connection begin...")); err != nil {
        
    		fmt.Println("err")
    	}
    }
    
    //连接断开之前要执行的逻辑
    func DoConnLost(conn ziface.IConnection) {
        
    	fmt.Println("=====>Do Conn Lost...")
    	fmt.Println("connID=", conn.GetConnectionID(), " is Lost....")
    }
    
    func main() {
        
    	s := znet.NewServer("[Zinx v0.9]")
    	//添加自定义路由（PingRouter和HelloRouter）
    	router0 := &PingRouter{
        }
    	s.AddRouter(0, router0)
    	router1 := &HelloRouter{
        }
    	s.AddRouter(1, router1)
    	//注册hook钩子函数
    	s.SetOnConnStart(DoConnBegin)
    	s.SetOnConnStop(DoConnLost)
    	s.Serve()
    }

> 测试代码中的myDemo/ZinxV0.9/Client.go和myDemo/ZinxV0.8/Client.go一样。

>  *  为了方便测试超过最大连接数的报错信息，我们可以修改配置文件  
>     ![在这里插入图片描述][3a582fc7aabe4593ad96a472d3e37179.png]

//将最大连接数设置为2，然后我们复制Client.go，可以多起几个Client来进行测试
    {
        
      "Name": "Zinx Server Application",
      "Version": "V0.9",
      "Host": "0.0.0.0",
      "TcpPort": 8092,
      "MaxConn": 2,
      "MaxPackageSize": 1024,
      "WorkerPoolSize": 10
    }

> 测试最大连接数与连接管理：  
> ![在这里插入图片描述][69c7a42d0abf4ba08d5abe963f5950ae.png]

> 测试钩子函数：  
> ![在这里插入图片描述][b5ed1c1d481547129a164d02dde12fe5.png]

### 4 添加连接属性并测试【v0.10】 ###

> 通过map\[string\]interface\{\}来存储连接的属性值,通过RWLock来保证读写connection属性值安全
> 
>  *  设置连接属性
>  *  获取连接属性
>  *  移除连接属性

##### zinx/ziface/iconnection.go #####

package ziface
    
    import "net"
    
    type IConnection interface {
        
    	//启动连接
    	Start()
    	//停止连接
    	Stop()
    	//获取当前连接的Conn对象
    	GetTCPConnection() *net.TCPConn
    	//获取当前连接模块的id
    	GetConnectionID() uint32
    	//获取远程客户端的TCP状态 IP：Port
    	RemoteAddr() net.Addr
    	//发送数据
    	SendMsg(msgId uint32, data []byte) error
    	SetProperty(key string, value interface{
        })
    	GetProperty(key string) (interface{
        }, error)
    	RemoveProperty(key string)
    }
    
    //定义一个处理连接业务的方法
    type HandleFunc func(*net.TCPConn, []byte, int) error

##### zinx/znet/connection.go #####

package znet
    
    import (
    	"fmt"
    	"github.com/kataras/iris/v12/x/errors"
    	"io"
    	"myTest/zinx/util"
    	"myTest/zinx/ziface"
    	"net"
    	"sync"
    )
    
    type Connection struct {
        
    	Conn       *net.TCPConn
    	ConnID     uint32
    	isClosed   bool
    	msgChannel chan []byte
    	//告知当前的连接已经退出/停止（由Reader告知writer退出）
    	ExitChan     chan bool
    	MsgHandler   *MsgHandle
    	TcpServer    ziface.IServer
    	property     map[string]interface{
        }
    	propertyLock sync.RWMutex
    }
    
    func NewConnection(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandle *MsgHandle) *Connection {
        
    	c := &Connection{
        
    		Conn:       conn,
    		ConnID:     connID,
    		MsgHandler: msgHandle,
    		isClosed:   false,
    		msgChannel: make(chan []byte),
    		ExitChan:   make(chan bool, 1),
    		TcpServer:  server,
    		property: make(map[string]interface{
        }),
    	}
    	//将conn添加到connMgr中
    	c.TcpServer.GetConnMgr().Add(c)
    	return c
    }
    
    func (c *Connection) StartWriter() {
        
    	fmt.Println("[Writer Goroutine is running]")
    	defer fmt.Println("[conn Writer  goroutine exit!]", c.RemoteAddr().String())
    	//不断的阻塞等待channel的消息，然后将channel中的消息写给客户端
    	for {
        
    		select {
        
    		case data := <-c.msgChannel:
    			//有数据写给客户端
    			if _, err := c.Conn.Write(data); err != nil {
        
    				fmt.Println("Send data error , ", err)
    				return
    			}
    		case <-c.ExitChan:
    			//代表reader已经退出，此时writer也需要退出
    			return
    		}
    	}
    }
    
    func (c *Connection) StartReader() {
        
    	fmt.Println("reader goroutine is running...")
    	defer fmt.Println("[Reader goroutine is exit] connID=", c.ConnID, " remote addr is ", c.RemoteAddr().String())
    	defer c.Stop()
    	//读取数据
    	for {
        
    		//创建一个拆包对象
    		dp := NewDataPack()
    		//读取客户端的msg Head 二进制流 8字节
    		headData := make([]byte, dp.GetHeadLen())
    		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
        
    			fmt.Println("read msg head err ", err)
    			break
    		}
    		//拆包，将读取到的headData封装为msg
    		msg, err := dp.UnPack(headData)
    		if err != nil {
        
    			fmt.Println("unpack msg err ", err)
    			break
    		}
    		//根据dataLen，再次读取Data，放在msg.Data中，
    		var data []byte
    		//如果数据包中有数据，则读取
    		if msg.GetMsgLen() > 0 {
        
    			data = make([]byte, msg.GetMsgLen())
    			//将切片data读满
    			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
        
    				fmt.Println("read msg data err ", err)
    				break
    			}
    		}
    		msg.SetData(data)
    
    		//封装请求，改为router处理
    		r := Request{
        
    			conn: c,
    			msg:  msg,
    		}
    		//判断是否开启workerPool,如果没有开启则直接创建协程处理；如果开启则通过workerPool处理
    		if util.GlobalObject.WorkerPoolSize > 0 {
        
    			c.MsgHandler.SendMsgToTaskQueue(&r)
    		} else {
        
    			go c.MsgHandler.DoMsgHandler(&r)
    		}
    	}
    }
    
    //启动连接
    func (c *Connection) Start() {
        
    	fmt.Printf("ConnID %d is Start...", c.ConnID)
    	//开启读、写
    	go c.StartReader()
    	go c.StartWriter()
    	//执行钩子函数
    	c.TcpServer.CallOnConnStart(c)
    }
    
    //停止连接
    func (c *Connection) Stop() {
        
    	fmt.Println("Connection Stop()...ConnectionID = ", c.ConnID)
    	if c.isClosed {
        
    		return
    	}
    	c.isClosed = true
    	//连接关闭之前执行hook关闭的钩子函数
    	c.TcpServer.CallOnConnStop(c)
    	c.Conn.Close()
    	c.ExitChan <- true
    
    	//连接conn关闭时，需要从连接管理模块中移除
    	c.TcpServer.GetConnMgr().Remove(c)
    	close(c.msgChannel)
    	close(c.ExitChan)
    }
    
    //获取当前连接的Conn对象
    func (c *Connection) GetTCPConnection() *net.TCPConn {
        
    	return c.Conn
    }
    
    //获取当前连接模块的id
    func (c *Connection) GetConnectionID() uint32 {
        
    	return c.ConnID
    }
    
    //获取远程客户端的TCP状态 IP：Port
    func (c *Connection) RemoteAddr() net.Addr {
        
    	return c.Conn.RemoteAddr()
    }
    
    //发送数据
    func (c *Connection) SendMsg(msgId uint32, data []byte) error {
        
    	if c.isClosed {
        
    		return errors.New("connection closed\n")
    	}
    	//将data进行封包
    	dp := NewDataPack()
    	binaryMsg, err := dp.Pack(NewMessage(msgId, data))
    	if err != nil {
        
    		fmt.Println("Pack error msg id=", msgId)
    		return errors.New("pack error msg")
    	}
    	//将数据发送给客户端
    	if _, err := c.Conn.Write(binaryMsg); err != nil {
        
    		fmt.Println("write msg id ", msgId, " error ", err)
    		return errors.New("conn write err ")
    	}
    	return nil
    }
    
    func (c *Connection) SetProperty(key string, value interface{
        }) {
        
    	c.propertyLock.Lock()
    	defer c.propertyLock.Unlock()
    	c.property[key] = value
    }
    
    func (c *Connection) GetProperty(key string) (interface{
        }, error) {
        
    	c.propertyLock.RLock()
    	defer c.propertyLock.RUnlock()
    	if value, ok := c.property[key]; ok {
        
    		return value, nil
    	} else {
        
    		return nil, errors.New("no property found")
    	}
    }
    
    func (c *Connection) RemoveProperty(key string) {
        
    	c.propertyLock.Lock()
    	defer c.propertyLock.Unlock()
    	delete(c.property, key)
    }

##### 测试 #####

###### myDemo/ZinxV0.10/Server.go ######

package main
    
    import (
    	"fmt"
    	"myTest/zinx/ziface"
    	"myTest/zinx/znet"
    )
    
    //自定义一个Router，测试路由功能
    type PingRouter struct {
        
    	znet.BaseRouter
    }
    
    func (pr *PingRouter) Handler(request ziface.IRequest) {
        
    	fmt.Println("call router handler...")
    	//先读取客户端数据，再回写ping...ping...ping...
    	fmt.Println("receive from client msgId=", request.GetMsgID(),
    		"data=", string(request.GetData()))
    
    	//回写ping
    	err := request.GetConnection().SendMsg(0, []byte("ping...ping...ping..."))
    	if err != nil {
        
    		fmt.Println(err)
    	}
    }
    
    //定义第二个Router
    type HelloRouter struct {
        
    	znet.BaseRouter
    }
    
    func (hr *HelloRouter) Handler(request ziface.IRequest) {
        
    	fmt.Println("receive from client msgId=", request.GetMsgID(),
    		"data=", string(request.GetData()))
    	err := request.GetConnection().SendMsg(1, []byte("hello zinx, I'm the other handler"))
    	if err != nil {
        
    		fmt.Println(err)
    	}
    }
    
    //连接创建成功之后需要执行的逻辑
    func DoConnBegin(conn ziface.IConnection) {
        
    	fmt.Println("=====>Do Conn Begin...")
    	if err := conn.SendMsg(202, []byte("do connection begin...")); err != nil {
        
    		fmt.Println("err")
    	}
    	//给conn设置属性
    	conn.SetProperty("Name", "ziyi")
    	conn.SetProperty("士兵突击", "https://www.bilibili.com/video/BV1Lk4y1N7tC/")
    }
    
    //连接断开之前要执行的逻辑
    func DoConnLost(conn ziface.IConnection) {
        
    	fmt.Println("=====>Do Conn Lost...")
    	fmt.Println("connID=", conn.GetConnectionID(), " is Lost....")
    	//读取属性
    	property, _ := conn.GetProperty("Name")
    	fmt.Println("Get Property Name=", property)
    	property, _ = conn.GetProperty("士兵突击")
    	fmt.Println("Get Property 士兵突击=", property)
    }
    
    func main() {
        
    	s := znet.NewServer("[Zinx v0.10]")
    	//添加自定义路由（PingRouter和HelloRouter）
    	router0 := &PingRouter{
        }
    	s.AddRouter(0, router0)
    	router1 := &HelloRouter{
        }
    	s.AddRouter(1, router1)
    	//注册hook钩子函数
    	s.SetOnConnStart(DoConnBegin)
    	s.SetOnConnStop(DoConnLost)
    	s.Serve()
    }

###### myDemo/ZinxV0.10/Client.go ######

![在这里插入图片描述][154b5e9b43ef4eb1a802940887543ec4.png]

[0ad78ed681b04114b375519aa7875724.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/4e39fcedd4854101ac62d080f26e8426.png
[964159f564d94508864ac0d9329bb324.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/1e3563a7ab444c6cb1fda57c66f79786.png
[45caa17774a0445ea4f5636bbc7041bc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/3f777e357407413c83e8739b34b9a110.png
[16d52470510e40beb25b75c5d2974695.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/25f67d595642448284b475592a8ad70c.png
[9582aefc318f4cb489dab7813a5b6f64.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/e361bb1604094368af1521e65fcae8ae.png
[14098d4841c04c8aa488185a274d1f8a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/3f03a765903e41be8ff26c361b0d0e72.png
[14c33b84249142219365dd1f3f65d3a3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/6bb6440f9be94f50ac6b08c041cf5054.png
[3a582fc7aabe4593ad96a472d3e37179.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/021feaeaa27b4d1eab3c18eb5c41126a.png
[69c7a42d0abf4ba08d5abe963f5950ae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/2552dca0baf948599d5469a79bb6a9ca.png
[b5ed1c1d481547129a164d02dde12fe5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/8035120f82ed4e00af1be97afadaf016.png
[154b5e9b43ef4eb1a802940887543ec4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/c9f96ae6db6944e38b261a9a1eac0bb9.png