一 TCP粘包优雅处理

1.1 粘包产生的原因

在使用TCP协议分高发送数据时，为了提高发送效率，需要对封包进行拼包处理，将小包凑成大包，在TCP层可以节约包头的大小损耗，I/O层的调用损耗也可以有所降低。但此时也容易形成粘包，也就是没有按照预期的大小得到数据，数据包不完整。

在接收TCP封包时，接收缓冲区的大小与发送过来的TCP传输单元大小不等，这时候会造成两种情况：

• 接收的数据大于等于接收缓冲区大小时，此时需要将数据复制到用户缓冲，接着读取后面的封包。
• 接收的数据小于接收缓冲区大小时，此时需要继续等待后续的 TCP 封包。

在go语言的io包中有个函数ReadAtLeast()用来处理封装

func ReadAtLeast(r Reader, buf []byte, min int) (n int, err error) {
    if len(buf) < min {
        return 0, ErrShortBuffer
    }
    for n < min && err == nil {
        var nn int
        nn, err = r.Read(buf[n:])
        n += nn
    }
    if n >= min {
        err = nil
    } else if n > 0 && err == EOF {
        err = ErrUnexpectedEOF
    }
    return
}

在else判断中，读取目标已经完结，但是己经读取一些数据，也就是说，没法完成读取任务，发生了不可期望的终结错误。

ReadAtLeast还有个更好的封装：

func ReadFull(r Reader, buf []byte, min int) (n int, err error) {
    return ReadAtLeast(r, buf, len(buf))
}

这个函数只需要提供buf接收缓冲区切片，就可以将这个己经分配的buf填充满 。简单地说就是: 给多大空间，填充多少字节，直到填满空间。
使用ReadFull可以优雅地完成对TCP粘包的处理。

1.2 封包发送

封包格式：

• Size：大小，代表Size后面包体的大小
• Body：消息的二进制数据

在发送封包时，需要将封包的Size字段从无符号十六位整型转换为字节数组（利用binary包的Write函数）：

// 二进制封包格式
type Packet struct {
    Size uint16 // 包体大小
    Body []byte // 包体数据
}

// 将数据写入dataWriter
func writePacket(dataWriter io.Writer, data []byte) error {

    // 准备一个字节数组缓冲
    var buf bytes.Buffer

    // 将Size写入缓冲
    err := binary.Write(&buf, binary.LittleEndian, uint16(len(data)))
    if err != nil {
        return err
    }

    // 写入包体数据
    _, err = buf.Write(data)
    if err != nil {
        return err
    }

    // 获取写入的完整数据
    out := buf.Bytes()

    // 写入完整数据
    _, err = dataWriter.Write(out)
    if err != nil {
        return err
    }

    return nil
}

1.3 连接器

连接器可以通过给定的地址和发送次数，不断地通过Socket给地址对应的连接发送封包。本例中，将通过循环构造的转为字符串的数字形成封包。由于速度快，包量大，因此能有效地造成粘包：

func Connector(addr string, sendTimes int) {

    // 创建连接
    conn, err := net.Dial("tpc", addr)
    if err != nil {
        fmt.Println("连接错误：", err)
        return
    }

    // 循环请求
    for i := 0; i < sendTimes; i++ {
        str := strconv.Itoa(i) // 将循环序号转化为字符串
        err := writePacket(conn, []byte(str))
        if err != nil {
            fmt.Println("发送错误：", err)
            break
        }
    }
}

1.4 接受器

连接器( Connector)只能发起到接收器( Acceptor)的连接， 一个接受器能接受多个来源连接。

接受器中侦昕的过程被放到一个独立的goroutine中，保证侦昕的过程不会阻塞其他逻辑的执行。在侦昕器停止时，需要使用 sync.WaitGroup进行同步，确认侦昕过程正常结 束。接受器不停的接受连接，当获取到连接时，为连接创建一个并发的会话处理的goroutine( handleSession 函数)。

接受器开始侦昕后，使用 Wait()方法等待侦昕结束。当处理粘包的任务完成时俗，Acceptor的Stop()方法会被调用，结束侦昕，同时退出整个程序。接受器的实现过程如下：

// 接收器
type Acceptor struct {
    ls            net.Listener   // 保存侦听器
    wg            sync.WaitGroup // 侦听器同步
    OnSeesionData func(net.Conn, []byte) bool
}

// 异步开始侦听
func (a *Acceptor) Start(addr string) {
    go func(addr string) {
        a.wg.Add(1) // 侦听开始时，添加一个任务
        defer a.wg.Done()

        var err error
        a.ls, err = net.Listen("tcp", addr)
        if err != nil {
            fmt.Println("listen err = ", err)
            return
        }
        // 侦听循环
        for {
            conn, err := a.ls.Accept() // 新连接没到来时，Accept阻塞
            if err != nil {
                break
            }
            go handleSession(conn, a.OnSeesionData)
        }
    }(addr)
}

// 停止侦听
func (a *Acceptor) Stop() {
    a.ls.Close()
}

// 等待侦听完全停止
func (a *Acceptor) Wait() {
    a.wg.Wait()
}

func NewAcceptor() *Acceptor {
    return &Acceptor{}
}

1.5 封包读取

本例中的 readPacket（）函数可以从 dataReader 中读取封包数据，并且将封包的 Size 和 Body 部分分离并返回 Packet 结构：

func readPacket(dataReader io.Reader) (pkt Packet, err error) {
    var sizeBuffer = make([]byte, 2) // Size为uint16类型，占据2个字节

    // 持续读取Size，直到读到为止
    _, err = io.ReadFull(dataReader, sizeBuffer)
    if err != nil {
        return
    }

    // 读取数据
    sizeReader := bytes.NewReader(sizeBuffer)
    err = binary.Read(sizeReader, binary.LittleEndian, &pkt.Size)
    if err != nil {
        return
    }

    // 分配包体大小
    pkt.Body = make([]byte, pkt.Size)

    // 读取包体数据
    _, err = io.ReadFull(dataReader, pkt.Body)
    return
}

1.6 会话处理

务器在接受一个连接后就会进入会话（ Session ）处理。会话处理是一个循环，不停地从 Socket 中接收数据并通过 readPacket（）函数将数据转换为封包。会话处理会传入一个处理数据的回调，处理好的数据会通过这个回调传送到用户的逻辑回调函数（ callback) ：

// 连接的会话逻辑
func handleSession(conn net.Conn, callback func(net.Conn, []byte) bool) {
    dataReader := bufio.NewReader(conn) // 创建socket读取器
    for {                               // 循环接收数据
        pkt, err := readPacket(dataReader)
        if err != nil || !callback(conn, pkt.Body) {
            // 回调要求退出
            conn.Close()
            break
        }
    }
}

1.5 测试粘包处理

func main() {

    const TESTCOUNT = 10000 // 测试次数
    const addr = "127.0.0.1:3000"

    var recCounter int // 接收器

    a := NewAcceptor()
    a.Start(addr)

    a.OnSeesionData = func(conn net.Conn, data []byte) bool {
        str := string(data)
        n, err := strconv.Atoi(str)
        if err != nil || recCounter != n {
            panic("failed")
        }
        recCounter++

        if recCounter >= TESTCOUNT {
            a.Stop()
            return false
        }

        return true
    }

    // 连接器不断发送数据
    Connector(addr, TESTCOUNT)

    // 等待侦听器结束
    a.Wait()
}