我们这里就是采用经典的TLV(Type-Len-Value)封包格式来解决TCP粘包问题吧。

TCP粘包问题-拆包封包过程
由于Zinx也是TCP流的形式传播数据,难免会出现消息1和消息2一同发送,那么zinx就需要有能力区分两个消息的边界,所以Zinx此时应该提供一个统一的拆包和封包的方法。在发包之前打包成如上图这种格式的有head和body的两部分的包,在收到数据的时候分两次进行读取,先读取固定长度的head部分,得到后续Data的长度,再根据DataLen读取之后的body。这样就能够解决粘包的问题了。

A) 创建拆包封包抽象类

zinx/ziface下,创建idatapack.go文件

zinx/ziface/idatapack.go

  1. package ziface
  2. /*
  3. 封包数据和拆包数据
  4. 直接面向TCP连接中的数据流,为传输数据添加头部信息,用于处理TCP粘包问题。
  5. */
  6. type IDataPack interface{
  7. GetHeadLen() uint32 //获取包头长度方法
  8. Pack(msg IMessage)([]byte, error) //封包方法
  9. Unpack([]byte)(IMessage, error) //拆包方法
  10. }

B) 实现拆包封包类

zinx/znet/下,创建datapack.go文件.

zinx/znet/datapack.go

  1. package znet
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "errors"
  6. "zinx/utils"
  7. "zinx/ziface"
  8. )
  9. //封包拆包类实例,暂时不需要成员
  10. type DataPack struct {}
  11. //封包拆包实例初始化方法
  12. func NewDataPack() *DataPack {
  13. return &DataPack{}
  14. }
  15. //获取包头长度方法
  16. func(dp *DataPack) GetHeadLen() uint32 {
  17. //Id uint32(4字节) + DataLen uint32(4字节)
  18. return 8
  19. }
  20. //封包方法(压缩数据)
  21. func(dp *DataPack) Pack(msg ziface.IMessage)([]byte, error) {
  22. //创建一个存放bytes字节的缓冲
  23. dataBuff := bytes.NewBuffer([]byte{})
  24. //写dataLen
  25. if err := binary.Write(dataBuff, binary.LittleEndian, msg.GetDataLen()); err != nil {
  26. return nil, err
  27. }
  28. //写msgID
  29. if err := binary.Write(dataBuff, binary.LittleEndian, msg.GetMsgId()); err != nil {
  30. return nil, err
  31. }
  32. //写data数据
  33. if err := binary.Write(dataBuff, binary.LittleEndian, msg.GetData()); err != nil {
  34. return nil ,err
  35. }
  36. return dataBuff.Bytes(), nil
  37. }
  38. //拆包方法(解压数据)
  39. func(dp *DataPack) Unpack(binaryData []byte)(ziface.IMessage, error) {
  40. //创建一个从输入二进制数据的ioReader
  41. dataBuff := bytes.NewReader(binaryData)
  42. //只解压head的信息,得到dataLen和msgID
  43. msg := &Message{}
  44. //读dataLen
  45. if err := binary.Read(dataBuff, binary.LittleEndian, &msg.DataLen); err != nil {
  46. return nil, err
  47. }
  48. //读msgID
  49. if err := binary.Read(dataBuff, binary.LittleEndian, &msg.Id); err != nil {
  50. return nil, err
  51. }
  52. //判断dataLen的长度是否超出我们允许的最大包长度
  53. if (utils.GlobalObject.MaxPacketSize > 0 && msg.DataLen > utils.GlobalObject.MaxPacketSize) {
  54. return nil, errors.New("Too large msg data recieved")
  55. }
  56. //这里只需要把head的数据拆包出来就可以了,然后再通过head的长度,再从conn读取一次数据
  57. return msg, nil
  58. }

需要注意的是整理的Unpack方法,因为我们从上图可以知道,我们进行拆包的时候是分两次过程的,第二次是依赖第一次的dataLen结果,所以Unpack只能解压出包头head的内容,得到msgId 和 dataLen。之后调用者再根据dataLen继续从io流中读取body中的数据。

C) 测试拆包封包功能

为了容易理解,我们先不用集成zinx框架来测试,而是单独写一个Server和Client来测试一下封包拆包的功能

Server.go

  1. package main
  2. import (
  3. "fmt"
  4. "io"
  5. "net"
  6. "zinx/znet"
  7. )
  8. //只是负责测试datapack拆包,封包功能
  9. func main() {
  10. //创建socket TCP Server
  11. listener, err := net.Listen("tcp", "127.0.0.1:7777")
  12. if err != nil {
  13. fmt.Println("server listen err:", err)
  14. return
  15. }
  16. //创建服务器gotoutine,负责从客户端goroutine读取粘包的数据,然后进行解析
  17. for {
  18. conn, err := listener.Accept()
  19. if err != nil {
  20. fmt.Println("server accept err:", err)
  21. }
  22. //处理客户端请求
  23. go func(conn net.Conn) {
  24. //创建封包拆包对象dp
  25. dp := znet.NewDataPack()
  26. for {
  27. //1 先读出流中的head部分
  28. headData := make([]byte, dp.GetHeadLen())
  29. _, err := io.ReadFull(conn, headData) //ReadFull 会把msg填充满为止
  30. if err != nil {
  31. fmt.Println("read head error")
  32. break
  33. }
  34. //将headData字节流 拆包到msg中
  35. msgHead, err := dp.Unpack(headData)
  36. if err != nil {
  37. fmt.Println("server unpack err:", err)
  38. return
  39. }
  40. if msgHead.GetDataLen() > 0 {
  41. //msg 是有data数据的,需要再次读取data数据
  42. msg := msgHead.(*znet.Message)
  43. msg.Data = make([]byte, msg.GetDataLen())
  44. //根据dataLen从io中读取字节流
  45. _, err := io.ReadFull(conn, msg.Data)
  46. if err != nil {
  47. fmt.Println("server unpack data err:", err)
  48. return
  49. }
  50. fmt.Println("==> Recv Msg: ID=", msg.Id, ", len=", msg.DataLen, ", data=", string(msg.Data))
  51. }
  52. }
  53. }(conn)
  54. }
  55. }

Client.go

  1. package main
  2. import (
  3. "fmt"
  4. "net"
  5. "zinx/znet"
  6. )
  7. func main() {
  8. //客户端goroutine,负责模拟粘包的数据,然后进行发送
  9. conn, err := net.Dial("tcp", "127.0.0.1:7777")
  10. if err != nil {
  11. fmt.Println("client dial err:", err)
  12. return
  13. }
  14. //创建一个封包对象 dp
  15. dp := znet.NewDataPack()
  16. //封装一个msg1包
  17. msg1 := &znet.Message{
  18. Id: 0,
  19. DataLen: 5,
  20. Data: []byte{'h', 'e', 'l', 'l', 'o'},
  21. }
  22. sendData1, err := dp.Pack(msg1)
  23. if err != nil {
  24. fmt.Println("client pack msg1 err:", err)
  25. return
  26. }
  27. msg2 := &znet.Message{
  28. Id: 1,
  29. DataLen: 7,
  30. Data: []byte{'w', 'o', 'r', 'l', 'd', '!', '!'},
  31. }
  32. sendData2, err := dp.Pack(msg2)
  33. if err != nil {
  34. fmt.Println("client temp msg2 err:", err)
  35. return
  36. }
  37. //将sendData1,和 sendData2 拼接一起,组成粘包
  38. sendData1 = append(sendData1, sendData2...)
  39. //向服务器端写数据
  40. conn.Write(sendData1)
  41. //客户端阻塞
  42. select {}
  43. }

运行Server.go

  1. go run Server.go

运行Client.go

  1. go run Client.go

我们从服务端看到运行结果

  1. $go run Server.go
  2. ==> Recv Msg: ID= 0 , len= 5 , data= hello
  3. ==> Recv Msg: ID= 1 , len= 7 , data= world!!

我们成功的得到了客户端发送的两个包,并且成功的解析出来。