Ainx的消息封装

📕作者简介： 过去日记，致力于Java、GoLang,Rust等多种编程语言，热爱技术，喜欢游戏的博主。
📗本文收录于Ainx系列，大家有兴趣的可以看一看
📘相关专栏Rust初阶教程、go语言基础系列、spring教程等，大家有兴趣的可以看一看
📙Java并发编程系列，设计模式系列、go web开发框架系列正在发展中，喜欢Java，GoLang，Rust，的朋友们可以关注一下哦！

📙 本文大部分都是借鉴刘丹冰大佬的zinx框架和文章，更推荐大家去读大佬的原文,本文只是个人学习的记录

文章目录

Ainx的消息封装
- 创建消息封装类型
- 消息的封包与拆包
- - 创建拆包封包抽象类
  - 实现拆包封包类
- Ainx-V0.5代码实现
- - Request字段修改
  - 集成拆包过程
  - 提供封包方法
- 使用Ainx-V0.5完成应用程序

Ainx的消息封装

接下来我们再对Ainx做一个简单的升级，现在我们把服务器的全部数据都放在一个Request里，当前的Request结构如下：

type Request struct {conn ziface.IConnection //已经和客户端建立好的链接data []byte             //客户端请求的数据
}

很明显，现在是用一个[]byte来接受全部数据，又没有长度，又没有消息类型，这不科学。怎么办呢？我们现在就要自定义一种消息类型，把全部的消息都放在这种消息类型里。

创建消息封装类型

在ainx/ainterface/下创建imessage.go文件

ainx/ainterface/imessage.go

package ainterface/*
将请求的一个消息封装到message中，定义抽象层接口
*/
type IMessage interface {GetDataLen() uint32 //获取消息数据段长度GetMsgId() uint32   //获取消息IDGetData() []byte    //获取消息内容SetMsgId(uint32)   //设计消息IDSetData([]byte)    //设计消息内容SetDataLen(uint32) //设置消息数据段长度
}

同时创建实例message类，在ainx/anet/下，创建message.go文件

ainx/anet/message.go

package anettype Message struct {Id      uint32 //消息的IDDataLen uint32 //消息的长度Data    []byte //消息的内容
}// 创建一个Message消息包
func NewMsgPackage(id uint32, data []byte) *Message {return &Message{Id:      id,DataLen: uint32(len(data)),Data:    data,}
}// 获取消息数据段长度
func (msg *Message) GetDataLen() uint32 {return msg.DataLen
}// 获取消息ID
func (msg *Message) GetMsgId() uint32 {return msg.Id
}// 获取消息内容
func (msg *Message) GetData() []byte {return msg.Data
}// 设置消息数据段长度
func (msg *Message) SetDataLen(len uint32) {msg.DataLen = len
}// 设计消息ID
func (msg *Message) SetMsgId(msgId uint32) {msg.Id = msgId
}// 设计消息内容
func (msg *Message) SetData(data []byte) {msg.Data = data
}

整理一个基本的message包，会包含消息ID，数据，数据长度三个成员，提供基本的setter和getter方法，目的是为了以后做封装优化的作用。同时也提供了一个创建一个message包的初始化方法NewMegPackage。

消息的封包与拆包

我们这里就是采用经典的TLV(Type-Len-Value)封包格式来解决TCP粘包问题吧。
外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传
由于Ainx也是TCP流的形式传播数据，难免会出现消息1和消息2一同发送，那么zinx就需要有能力区分两个消息的边界，所以Ainx此时应该提供一个统一的拆包和封包的方法。在发包之前打包成如上图这种格式的有head和body的两部分的包，在收到数据的时候分两次进行读取，先读取固定长度的head部分，得到后续Data的长度，再根据DataLen读取之后的body。这样就能够解决粘包的问题了。

创建拆包封包抽象类

在ainx/ainterface下，创建idatapack.go文件
ainterface

ainx/ainterface/idatapack.go

package ainterface/*
封包数据和拆包数据
直接面向TCP链接中的数据流，为传输数据添加头部信息，用于处理TCP粘包问题。
*/
type IDataPack interface {GetHeadLen() uint32                //获取包头长度方法Pack(msg IMessage) ([]byte, error) //封包方法Unpack([]byte) (IMessage, error)   //拆包方法
}

实现拆包封包类

在ainx/anet/下，创建datapack.go文件.

ainx/anet/datapack.go

package anetimport ("ainx/ainterface""ainx/utils""bytes""encoding/binary""errors"
)// 封包拆包实例，暂时不需要成员
type DataPack struct {
}// 封包拆包实例初始化方法
func NewDataPack() *DataPack {return &DataPack{}
}// 获取包头长度方法
func (dp *DataPack) GetHeadLen() uint32 {//Id uint32(4字节) +  DataLen uint32(4字节)return 8
}// 封包方法(压缩)
func (dp *DataPack) Pack(msg ainterface.IMessage) ([]byte, error) {// 创建一个存放bytes字节的缓冲dataBuff := bytes.NewBuffer([]byte{})写dataLen//字节序 就是多字节数据类型 (int, float 等)在内存中的存储顺序。可分为大端序，低地址端存放高位字节；小端序与之相反，低地址端存放低位字节。if err := binary.Write(dataBuff, binary.LittleEndian, msg.GetDataLen()); err != nil {return nil, err}//写msgIDif err := binary.Write(dataBuff, binary.LittleEndian, msg.GetMsgId()); err != nil {return nil, err}//写data数据if err := binary.Write(dataBuff, binary.LittleEndian, msg.GetData()); err != nil {return nil, err}return dataBuff.Bytes(), nil
}// 拆包方法(解压数据)
func (dp *DataPack) Unpack(binaryData []byte) (ainterface.IMessage, error) {//创建一个从输入二进制数据的ioReaderdataBuff := bytes.NewReader(binaryData)//只解压head的信息，得到dataLen和msgIDmsg := &Message{}//读dataLenif err := binary.Read(dataBuff, binary.LittleEndian, &msg.DataLen); err != nil {return nil, err}//读msgIDif err := binary.Read(dataBuff, binary.LittleEndian, &msg.Id); err != nil {return nil, err}//判断dataLen的长度是否超出我们允许的最大包长度if utils.GlobalSetting.MaxPacketSize > 0 && msg.DataLen > utils.GlobalSetting.MaxPacketSize {return nil, errors.New("Too large msg data recieved")}//这里只需要把head的数据拆包出来就可以了，然后再通过head的长度，再从conn读取一次数据return msg, nil
}

Ainx-V0.5代码实现

现在我们需要把封包和拆包的功能集成到Zinx中，并且测试Zinx该功能是否生效。

Request字段修改

首先我们要将我们之前的Request中的[]byte类型的data字段改成Message类型.

ainx/anet/request.go

package anetimport "ainx/ainterface"type Request struct {conn ainterface.IConnection //已经和客户端建立好的链接msg  ainterface.IMessage    //客户端请求数据
}// 获取请求链接信息
func (r *Request) GetConnection() ainterface.IConnection {return r.conn
}// 获取请求消息的数据
func (r *Request) GetData() []byte {return r.msg.GetData()
}// 获取请求的消息的ID
func (r *Request) GetMsgID() uint32 {return r.msg.GetMsgId()
}

集成拆包过程

接下来我们需要在Connection的StartReader()方法中,修改之前的读取客户端的这段代码：

func (c *Connection) StartReader() {//...for  {//读取我们最大的数据到buf中buf := make([]byte, utils.GlobalSetting.MaxPacketSize)_, err := c.Conn.Read(buf)if err != nil {fmt.Println("recv buf err ", err)c.ExitBuffChan <- truecontinue}//...}
}

改成如下：

ainx/anet/connection.go

StartReader()方法

// 处理conn读数据的Goroutine
func (c *Connection) StartReader() {fmt.Println("Reader Goroutine is  running")defer fmt.Println(c.RemoteAddr().String(), " conn reader exit!")defer c.Stop()for {// 创建拆包解包的对象dp := NewDataPack()//读取客户端的Msg headheadData := make([]byte, dp.GetHeadLen())if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {fmt.Println("read msg head error ", err)c.ExitBuffChan <- truecontinue}//拆包，得到msgid 和 datalen 放在msg中msg, err := dp.Unpack(headData)if err != nil {fmt.Println("unpack error ", err)c.ExitBuffChan <- truecontinue}//根据 dataLen 读取 data，放在msg.Data中var data []byteif msg.GetDataLen() > 0 {data = make([]byte, msg.GetDataLen())if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {fmt.Println("read msg data error ", err)c.ExitBuffChan <- truecontinue}}msg.SetData(data)//得到当前客户端请求的Request数据req := Request{conn: c,msg:  msg, //将之前的buf 改成 msg}//从路由Routers 中找到注册绑定Conn的对应Handlego func(request ainterface.IRequest) {//执行注册的路由方法c.Router.PreHandle(request)c.Router.Handle(request)c.Router.PostHandle(request)}(&req)}
}

提供封包方法

现在我们已经将拆包的功能集成到Ainx中了，但是使用Ainx的时候，如果我们希望给用户返回一个TLV格式的数据，总不能每次都经过这么繁琐的过程，所以我们应该给Ainx提供一个封包的接口，供Ainx发包使用。

ainx/ainterface/iconnection.go

新增SendMsg()方法


type IConnection interface {// 启动连接，让当前连接开始工作Start()// 停止链接，结束当前连接状态Stop()//从当前连接获取原始的socket TCPConn GetTCPConnection() *net.TCPConn //获取当前连接IDGetConnID() uint32       //获取远程客户端地址信息 RemoteAddr() net.AddrGetConnection() net.Conn //  (从当前连接获取原始的socket TCPConn)//直接将Message数据发送数据给远程的TCP客户端SendMsg(msgId uint32, data []byte) error
}// 定义⼀一个统⼀一处理理链接业务的接⼝口
type HandFunc func(*net.TCPConn, []byte, int) error

ainx/anet/connection.go

SendMsg()方法实现:

// 直接将Message数据发送数据给远程的TCP客户端
func (c *Connection) SendMsg(msgId uint32, data []byte) error {if c.isClosed == true {return errors.New("Connection closed when send msg")}//将data封包，并且发送dp := NewDataPack()msg, err := dp.Pack(NewMsgPackage(msgId, data))if err != nil {fmt.Println("Pack error msg id = ", msgId)return errors.New("Pack error msg ")}//写回客户端if _, err := c.Conn.Write(msg); err != nil {fmt.Println("Write msg id ", msgId, " error ")c.ExitBuffChan <- truereturn errors.New("conn Write error")}return nil
}

使用Ainx-V0.5完成应用程序

现在我们可以基于Ainx框架完成发送msg功能的测试用例了。

Server.go

package mainimport ("ainx/ainterface""ainx/anet""fmt"
)// ping test 自定义路由
type PingRouter struct {anet.BaseRouter
}// Test Handle
func (this *PingRouter) Handle(request ainterface.IRequest) {fmt.Println("Call PingRouter Handle")//先读取客户端的数据，再回写ping...ping...pingfmt.Println("recv from client : msgId=", request.GetMsgID, ", data=", string(request.GetData()))//回写数据err := request.GetConnection().SendMsg(1, []byte("ping...ping...ping"))if err != nil {fmt.Println(err)}
}func main() {//创建一个server句柄s := anet.NewServer("Ainx V0.5")//配置路由s.AddRouter(&PingRouter{})//开启服务s.Serve()
}

Client.go

package mainimport ("ainx/anet""fmt""io""net""time"
)/*
模拟客户端
*/
func main() {fmt.Println("Client Test ... start")//3秒之后发起测试请求，给服务端开启服务的机会time.Sleep(3 * time.Second)conn, err := net.Dial("tcp", "127.0.0.1:8080")if err != nil {fmt.Println("client start err, exit!")return}for {//发封包message消息dp := anet.NewDataPack()msg, _ := dp.Pack(anet.NewMsgPackage(0, []byte("Ainx V0.5 Client 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.(*anet.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)}
}