网站制作和维护费用阿里大数据平台

Day5-Part4：Zinx 的连接属性设置

在 Zinx 当中，我们使用 Server 来开启服务并监听指定的端口，当接收到来自客户端的连接请求之后，Zinx 新建 Connection 来管理与 client 的连接，Connection 负责读取 client 发送给 Server 的数据（当然，需要拆包），通过 Router 进行业务处理，可能需要通过 Writer 将数据写回给 client，在这个过程中，client 可能有一些用户的数据或者参数，可以与 Connection 相绑定。

根据上述分析，我们现在将连接属性的绑定集成到 Zinx 框架当中。

给连接添加连接配置的接口

首先修改 iconnection.go 下的 IConnection 接口：

type IConnection interface {Start()                                      // 启动连接Stop()                                       // 停止连接GetConnID() uint32                           // 获取远程客户端地址信息GetTCPConnection() *net.TCPConn              // 从当前连接获取原始的 socket TCPConnRemoteAddr() net.Addr                        // 获取远程客户端地址信息SendMsg(msgId uint32, data []byte) error     // 直接将 Message 数据发给远程的 TCP 客户端SendBuffMsg(msgId uint32, data []byte) error // 添加带缓冲的发送消息接口SetProperty(key string, value interface{})   // 设置连接属性GetProperty(key string) (interface{}, error) // 获取连接属性RemoveProperty(key string)                   // 移除连接属性
}

新增了三个方法，分别是 SetProperty、GetProperty 和 RemoveProperty，分别用于添加、获取和移除属性。

连接属性方法的实现

现在将属性加入到 Connection 结构当中，并实现 IConnection 新增的方法：

type Connection struct {TCPServer    ziface.IServer    // 标记当前 Conn 属于哪个 ServerConn         *net.TCPConn      // 当前连接的 socket TCP 套接字ConnID       uint32            // 当前连接的 ID, 也可称为 SessionID, 全局唯一isClosed     bool              // 当前连接的开启/关闭状态Msghandler   ziface.IMsgHandle // 将 Router 替换为消息管理模块ExitBuffChan chan bool         // 告知该连接一经退出/停止的 channelmsgChan      chan []byte       // 无缓冲 channel, 用于读/写两个 goroutine 之间的消息通信msgBuffChan  chan []byte       // 定义 msgBuffChanproperty     map[string]interface{} // 连接属性propertyLock sync.RWMutex           // 保护连接属性修改的锁
}// NewConnection 创建新的连接
func NewConnection(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandler ziface.IMsgHandle) *Connection {c := &Connection{TCPServer:    server,Conn:         conn,ConnID:       connID,isClosed:     false,Msghandler:   msgHandler,ExitBuffChan: make(chan bool, 1),msgChan:      make(chan []byte), // msgChan 初始化msgBuffChan:  make(chan []byte, settings.Conf.MaxMsgChanLen),property:     make(map[string]interface{}),}// 将新创建的 Conn 添加到连接管理器中c.TCPServer.GetConnMgr().Add(c)return c
}// SetProperty 用于设置连接属性
func (c *Connection) SetProperty(key string, value interface{}) {c.propertyLock.Lock()defer c.propertyLock.Unlock()c.property[key] = value
}// GetProperty 获取连接属性
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")}
}// RemoveProperty 移除连接属性
func (c *Connection) RemoveProperty(key string) {c.propertyLock.Lock()defer c.propertyLock.Unlock()delete(c.property, key)
}

测试 Zinx-v1.0

现在我们正式完成了 Zinx-v1.0，开两个 main 函数，一个模拟 Server 一个模拟 Client，来对 Zinx 进行测试：

// zinx/server/main.go
package mainimport ("fmt""zinx/settings""zinx/ziface""zinx/znet"
)// ping test 自定义路由
type PingRouter struct {znet.BaseRouter
}// Ping Handle
func (this *PingRouter) Handle(request ziface.IRequest) {fmt.Println("Call PingRouter Handle")//先读取客户端的数据，再回写ping...ping...pingfmt.Println("recv from client : msgId=", request.GetMsgID(), ", data=", string(request.GetData()))err := request.GetConnection().SendBuffMsg(0, []byte("ping...ping...ping"))if err != nil {fmt.Println(err)}
}type HelloZinxRouter struct {znet.BaseRouter
}// HelloZinxRouter Handle
func (this *HelloZinxRouter) Handle(request ziface.IRequest) {fmt.Println("Call HelloZinxRouter Handle")//先读取客户端的数据，再回写ping...ping...pingfmt.Println("recv from client : msgId=", request.GetMsgID(), ", data=", string(request.GetData()))err := request.GetConnection().SendBuffMsg(1, []byte("Hello Zinx Router V0.10"))if err != nil {fmt.Println(err)}
}// 创建连接的时候执行
func DoConnectionBegin(conn ziface.IConnection) {fmt.Println("DoConnecionBegin is Called ... ")//=============设置两个链接属性，在连接创建之后===========fmt.Println("Set conn Name, Home done!")conn.SetProperty("Name", "yggp")conn.SetProperty("Home", "Learning Zinx")//===================================================err := conn.SendMsg(2, []byte("DoConnection BEGIN..."))if err != nil {fmt.Println(err)}
}// 连接断开的时候执行
func DoConnectionLost(conn ziface.IConnection) {//============在连接销毁之前，查询conn的Name，Home属性=====if name, err := conn.GetProperty("Name"); err == nil {fmt.Println("Conn Property Name = ", name)}if home, err := conn.GetProperty("Home"); err == nil {fmt.Println("Conn Property Home = ", home)}//===================================================fmt.Println("DoConnectionLost is Called ... ")
}func main() {//创建一个server句柄err := settings.Init()if err != nil {panic(err)}s := znet.NewServer()//注册链接hook回调函数s.SetOnConnStart(DoConnectionBegin)s.SetOnConnStop(DoConnectionLost)//配置路由s.AddRouter(0, &PingRouter{})s.AddRouter(1, &HelloZinxRouter{})//开启服务s.Serve()
}

// zinx/client/main.go
package mainimport ("fmt""io""net""time""zinx/znet"
)/*
模拟客户端
*/
func main() {fmt.Println("Client Test ... start")//3秒之后发起测试请求，给服务端开启服务的机会time.Sleep(3 * time.Second)conn, err := net.Dial("tcp", "127.0.0.1:7777")if err != nil {fmt.Println("client start err, exit!")return}for {//发封包message消息dp := znet.NewDataPack()msg, _ := dp.Pack(znet.NewMsgPackage(1, []byte("Zinx V1.0 Client1 Test Message")))_, err := conn.Write(msg)if err != nil {fmt.Println("write error err ", err)return}//先读出流中的head部分headData := make([]byte, dp.GetHeadLen())_, err = io.ReadFull(conn, headData) //ReadFull 会把msg填充满为止if err != nil {fmt.Println("read head error")break}//将headData字节流 拆包到msg中msgHead, err := dp.Unpack(headData)if err != nil {fmt.Println("server unpack err:", err)return}if msgHead.GetDataLen() > 0 {//msg 是有data数据的，需要再次读取data数据msg := msgHead.(*znet.Message)msg.Data = make([]byte, msg.GetDataLen())//根据dataLen从io中读取字节流_, err := io.ReadFull(conn, msg.Data)if err != nil {fmt.Println("server unpack data err:", err)return}fmt.Println("==> Recv Msg: ID=", msg.Id, ", len=", msg.DataLen, ", data=", string(msg.Data))}time.Sleep(1 * time.Second)}
}

Server 的 main.go 在执行过程中，Terminal 显示：

Add api msgId =  0
Add Router succ! msgId =  0
Add api msgId =  1
Add Router succ! msgId =  1
[START] Server listenner at IP: 127.0.0.1, Port 7777, is starting
[Zinx] Version: v1.0, MaxConn: 3, MaxPacketSize: 0
Worker ID =  0  is started.
Worker ID =  4  is started.
Worker ID =  5  is started.
Worker ID =  1  is started.
Worker ID =  6  is started.
Worker ID =  2  is started.
Worker ID =  3  is started.
Worker ID =  7  is started.
Worker ID =  8  is started.
Worker ID =  9  is started.
start Zinx server   zinx server  succ, now listenning...
connection add to ConnManager successfully: conn num =  1
---> CallOnConnStart ...
DoConnecionBegin is Called ...
Set conn Name, Home done!
[Writer Goroutine is running]
Reader Goroutine is running
Add ConnID =  0  request msgID =  1 to workerID =  0
Call HelloZinxRouter Handle
recv from client : msgId= 1 , data= Zinx V1.0 Client1 Test Message
Add ConnID =  0  request msgID =  1 to workerID =  0
Call HelloZinxRouter Handle
recv from client : msgId= 1 , data= Zinx V1.0 Client1 Test Message
Add ConnID =  0  request msgID =  1 to workerID =  0
Call HelloZinxRouter Handle
recv from client : msgId= 1 , data= Zinx V1.0 Client1 Test Message
Add ConnID =  0  request msgID =  1 to workerID =  0
Call HelloZinxRouter Handle
recv from client : msgId= 1 , data= Zinx V1.0 Client1 Test Message
Add ConnID =  0  request msgID =  1 to workerID =  0
... ... ...

当客户端退出时，显示：

... ... ...
recv from client : msgId= 1 , data= Zinx V1.0 Client1 Test Message
read msg head error read tcp4 127.0.0.1:7777->127.0.0.1:56961: wsarecv: An existing connection was forcibly closed by the remote host.
Conn Stop()... ConnID =  0
---> CallOnConnStop ...
Conn Property Name =  yggp
Conn Property Home =  Learning Zinx
DoConnectionLost is Called ...
connection Remove connID =  0  successfully: conn num =  0
127.0.0.1:56961  conn reader exit !
127.0.0.1:56961 [conn Writer exit!]

证明 hook 函数被成功调用，且可以成功读取 Connection 保存的 property。

Client 的 main.go 在 Terminal 显示：

Client Test ... start
==> Recv Msg: ID= 2 , len= 21 , data= DoConnection BEGIN...
==> Recv Msg: ID= 1 , len= 23 , data= Hello Zinx Router V0.10
==> Recv Msg: ID= 1 , len= 23 , data= Hello Zinx Router V0.10
==> Recv Msg: ID= 1 , len= 23 , data= Hello Zinx Router V0.10
==> Recv Msg: ID= 1 , len= 23 , data= Hello Zinx Router V0.10
==> Recv Msg: ID= 1 , len= 23 , data= Hello Zinx Router V0.10
... ... ...

至此，我们已经成功完成了 Zinx 项目的基本框架。

总结

前前后后花了 5 天时间来完成 Zinx-v1.0 框架，收获颇丰。Zinx 是一个轻量级的面向长连接的并发服务器框架，v1.0 版本只是它的一个基本雏形，在具备所有功能的基础上，仍然可以进一步升级。明天我会从头到尾对 Zinx 框架进行总结，并根据目前 github 上最新的 Zinx 项目谈一谈我未来会进一步做哪些改动，欢迎持续关注。