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   //获取消息ID
	GetData() []byte    //获取消息内容

	SetMsgId(uint32)   //设计消息ID
	SetData([]byte)    //设计消息内容
	SetDataLen(uint32) //设置消息数据段长度
}

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

ainx/anet/message.go

package anet

type Message struct {
	Id      uint32 //消息的ID
	DataLen 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 anet

import (
	"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
	}
	//写msgID
	if 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) {
	//创建一个从输入二进制数据的ioReader
	dataBuff := bytes.NewReader(binaryData)

	//只解压head的信息，得到dataLen和msgID
	msg := &Message{}

	//读dataLen
	if err := binary.Read(dataBuff, binary.LittleEndian, &msg.DataLen); err != nil {
		return nil, err
	}

	//读msgID
	if 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 anet

import "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 <- true
			continue
		}
        
        //...
        
    }
}

改成如下：

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 head
		headData := make([]byte, dp.GetHeadLen())
		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
			fmt.Println("read msg head error ", err)
			c.ExitBuffChan <- true
			continue
		}

		//拆包，得到msgid 和 datalen 放在msg中
		msg, err := dp.Unpack(headData)
		if err != nil {
			fmt.Println("unpack error ", err)
			c.ExitBuffChan <- true
			continue
		}

		//根据 dataLen 读取 data，放在msg.Data中
		var data []byte
		if 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 <- true
				continue
			}
		}
		msg.SetData(data)

		//得到当前客户端请求的Request数据
		req := Request{
			conn: c,
			msg:  msg, //将之前的buf 改成 msg
		}
		//从路由Routers 中找到注册绑定Conn的对应Handle
		go 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 //获取当前连接ID
	GetConnID() uint32       //获取远程客户端地址信息 RemoteAddr() net.Addr
	GetConnection() 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 <- true
		return errors.New("conn Write error")
	}

	return nil
}

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

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

Server.go

package main

import (
	"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...ping
	fmt.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 main

import (
	"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)
	}
}