4. 设计与实现

4.1 建立对自定义应用协议的抽象

程序是对现实世界的抽象。对于现实世界的自定义应用协议规范，需要在程序世界建立起对这份协议的抽象。

在进行抽象之前，先建立这次实现要用的源码项目 tcp-server-demo1，建立的步骤如下：

$mkdir tcp-server-demo1
$cd tcp-server-demo1
$go mod init github.com/bigwhite/tcp-server-demo1
go: creating new go.mod: module github.com/bigwhite/tcp-server-demo1

4.2 深入协议字段

自定义应用协议：高度简化的、基于二进制模式定义的协议。

二进制模式定义的特点：采用长度字段标识独立数据包的边界。

在这个协议规范中可得：

• 请求包和应答包的第一个字段（totalLength）都是包的总长度，用来标识包边界的那个字段，也是在应用层用于“分割包”的最重要字段。

• 请求包与应答包的第二个字段也一样，都是 commandID，这个字段用于标识包类型，定义四种包类型：

  • 连接请求包（值为 0x01）
  • 消息请求包（值为 0x02）
  • 连接响应包（值为 0x81）
  • 消息响应包（值为 0x82）

  转换为对应的代码就是：

  const (
      CommandConn   = iota + 0x01 // 0x01，连接请求包
      CommandSubmit               // 0x02，消息请求包
  )
  
  const (
      CommandConnAck   = iota + 0x81 // 0x81，连接请求的响应包
      CommandSubmitAck               // 0x82，消息请求的响应包
  )
  

• 请求包与应答包的第三个字段都是 ID，ID 是每个连接上请求包的消息流水号，顺序累加，步长为 1，循环使用，多用来请求发送方后匹配响应包，所以要求一对请求与响应消息的流水号必须相同。

• 请求包与响应包唯一的不同之处，就在于最后一个字段：

  请求包定义了有效载荷（payload），这个字段承载了应用层需要的业务数据；

  响应包则定义了请求包的响应状态字段（result），这里其实简化了响应状态字段的取值，成功的响应用 0 表示，如果是失败的响应，无论失败原因是什么，都用 1 来表示。

4.3 建立 Frame 和 Packet 抽象

TCP 连接上的数据是一个没有边界的字节流，但在业务层眼中，没有字节流，只有各种协议消息。因此，无论是从客户端到服务端，还是从服务端到客户端，业务层在连接上看到的都应该是一个挨着一个的协议消息流。

建立第一个抽象：Frame。每个 Frame 表示一个协议消息，这样在业务层眼中，连接上的字节流就是由一个接着一个 Frame 组成的，如下图所示：

自定义协议就封装在这一个个的 Frame 中。协议规定了将 Frame 分割开来的方法，那就是利用每个 Frame 开始处的 totalLength，每个 Frame 由一个 totalLength 和 Frame 的负载（payload）构成，比如下图中左侧的 Frame 结构：

通过 Frame header: totalLength 就可以将 Frame 之间隔离开来。

建立第二个抽象：Packet。将 Frame payload 定义为一个 Packet。上图右侧展示的就是 Packet 的结构。

Packet 就是业务层真正需要的消息，每个 Packet 由 Packet 头和 Packet Body 部分组成。Packet 头就是 commandID，用于标识这个消息的类型；而 ID 和 payload（packet payload）或 result 字段组成了 Packet 的 Body 部分，对业务层有价值的数据都包含在 Packet Body 部分。

4.4 协议的解包与打包

基于 Frame 和 Packet 这两个概念，实现对私有协议的解包与打包操作：

• 协议的解包（decode），就是指识别 TCP 连接上的字节流，将一组字节“转换”成一个特定类型的协议消息结构，然后这个消息结构会被业务处理逻辑使用；
• 协议的打包（encode），刚刚好相反，是指将一个特定类型的消息结构转换为一组字节，然后这组字节数据会被放在连接上发送出去。

具体到自定义协议上，解包就是指 字节流 -> Frame，打包是指 Frame -> 字节流。针对这个协议的服务端解包与打包的流程图：

TCP 流数据先后经过 frame decode 和 packet decode，得到应用层所需的 packet 数据，而业务层回复的响应，则先后经过 packet 的 encode 与 frame 的 encode，写入 TCP 数据流中。

4.5 Frame 的实现

协议部分最重要的两个抽象是 Frame 和 Packet，在项目中建立 frame 包与 packet 包，分别与两个协议抽象对应。

frame 包的职责是提供识别 TCP 流边界的编解码器，使用 StreamFrameCodec 接口来做编解码操作。

StreamFrameCodec接口示例：

// tcp-server-demo1/frame/frame.go

type FramePayload []byte

type StreamFrameCodec interface {
 Encode(io.Writer, FramePayload) error   // data -> frame，并写入io.Writer
 Decode(io.Reader) (FramePayload, error) // 从io.Reader中提取frame payload，并返回给上层
}

StreamFrameCodec 接口类型有两个方法 Encode 与 Decode。

Encode 方法用于将输入的 Frame payload 编码为一个 Frame，然后写入 io.Writer 所代表的输出（outbound）TCP 流中。

Decode 方法正好相反，它从代表输入（inbound）TCP 流的 io.Reader 中读取一个完整 Frame，并将得到的 Frame payload 解析出来并返回。

定义出一个针对协议的统一的接口类型 StreamFrameCodec ：

// tcp-server-demo1/frame/frame.go

var ErrShortWrite = errors.New("short write")
var ErrShortRead = errors.New("short read")

type myFrameCodec struct{}

func NewMyFrameCodec() StreamFrameCodec {
    return &myFrameCodec{}
}

func (p *myFrameCodec) Encode(w io.Writer, framePayload FramePayload) error {
    var f = framePayload
    var totalLen int32 = int32(len(framePayload)) + 4

    err := binary.Write(w, binary.BigEndian, &totalLen)
    if err != nil {
        return err
    }

    n, err := w.Write([]byte(f)) // write the frame payload to outbound stream
    if err != nil {
        return err
    }
  
    if n != len(framePayload) {
        return ErrShortWrite
    }

    return nil
}

func (p *myFrameCodec) Decode(r io.Reader) (FramePayload, error) {
    var totalLen int32
    err := binary.Read(r, binary.BigEndian, &totalLen)
    if err != nil {
        return nil, err
    }

    buf := make([]byte, totalLen-4)
    n, err := io.ReadFull(r, buf)
    if err != nil {
        return nil, err
    }
  
    if n != int(totalLen-4) {
        return nil, ErrShortRead
    }

    return FramePayload(buf), nil
}

注意三点事项：

• 网络字节序使用大端字节序（BigEndian），因此无论是 Encode 还是 Decode，都是用 binary.BigEndian；
• binary.Read 或 Write 会根据参数的宽度，读取或写入对应的字节个数的字节，这里 totalLen 使用 int32，那么 Read 或 Write 只会操作数据流中的 4 个字节；
• 这里没有设置网络 I/O 操作的 Deadline，io.ReadFull 一般会读满你所需的字节数，除非遇到 EOF 或 ErrUnexpectedEOF。

在工程实践中，保证打包与解包正确的最有效方式就是编写单元测试，为 StreamFrameCodec 接口的实现编写测试用例。

myFrameCodec 的两个测试用例：

// tcp-server-demo1/frame/frame_test.go

func TestEncode(t *testing.T) {
    codec := NewMyFrameCodec()
    buf := make([]byte, 0, 128)
    rw := bytes.NewBuffer(buf)

    err := codec.Encode(rw, []byte("hello"))
    if err != nil {
        t.Errorf("want nil, actual %s", err.Error())
    }

    // 验证Encode的正确性
    var totalLen int32
    err = binary.Read(rw, binary.BigEndian, &totalLen)
    if err != nil {
        t.Errorf("want nil, actual %s", err.Error())
    }

    if totalLen != 9 {
        t.Errorf("want 9, actual %d", totalLen)
    }

    left := rw.Bytes()
    if string(left) != "hello" {
        t.Errorf("want hello, actual %s", string(left))
    }
}

func TestDecode(t *testing.T) {
    codec := NewMyFrameCodec()
    data := []byte{0x0, 0x0, 0x0, 0x9, 'h', 'e', 'l', 'l', 'o'}

    payload, err := codec.Decode(bytes.NewReader(data))
    if err != nil {
        t.Errorf("want nil, actual %s", err.Error())
    }

    if string(payload) != "hello" {
        t.Errorf("want hello, actual %s", string(payload))
    }
}

测试 Encode 方法，其实不需要建立真实的网络连接，只要用一个满足 io.Writer 的 bytes.Buffer 实例“冒充”真实网络连接就可以了，同时 bytes.Buffer 类型也实现了 io.Reader 接口，可以很方便地从中读取出 Encode 后的内容，并进行校验比对。

type ReturnErrorWriter struct {
    W  io.Writer
    Wn int // 第几次调用Write返回错误
    wc int // 写操作次数计数
}

func (w *ReturnErrorWriter) Write(p []byte) (n int, err error) {
    w.wc++
    if w.wc >= w.Wn {
        return 0, errors.New("write error")
    }
    return w.W.Write(p)
}

type ReturnErrorReader struct {
    R  io.Reader
    Rn int // 第几次调用Read返回错误
    rc int // 读操作次数计数
}

func (r *ReturnErrorReader) Read(p []byte) (n int, err error) {
    r.rc++
    if r.rc >= r.Rn {
        return 0, errors.New("read error")
    }
    return r.R.Read(p)
}

func TestEncodeWithWriteFail(t *testing.T) {
    codec := NewMyFrameCodec()
    buf := make([]byte, 0, 128)
    w := bytes.NewBuffer(buf)

    // 模拟binary.Write返回错误
    err := codec.Encode(&ReturnErrorWriter{
        W:  w,
        Wn: 1,
    }, []byte("hello"))
    if err == nil {
        t.Errorf("want non-nil, actual nil")
    }

    // 模拟w.Write返回错误
    err = codec.Encode(&ReturnErrorWriter{
        W:  w,
        Wn: 2,
    }, []byte("hello"))
    if err == nil {
        t.Errorf("want non-nil, actual nil")
    }
}

func TestDecodeWithReadFail(t *testing.T) {
    codec := NewMyFrameCodec()
    data := []byte{0x0, 0x0, 0x0, 0x9, 'h', 'e', 'l', 'l', 'o'}

    // 模拟binary.Read返回错误
    _, err := codec.Decode(&ReturnErrorReader{
        R:  bytes.NewReader(data),
        Rn: 1,
    })
    if err == nil {
        t.Errorf("want non-nil, actual nil")
    }

    // 模拟io.ReadFull返回错误
    _, err = codec.Decode(&ReturnErrorReader{
        R:  bytes.NewReader(data),
        Rn: 2,
    })
    if err == nil {
        t.Errorf("want non-nil, actual nil")
    }
}

为了提升测试覆盖率，需要尽可能让测试覆盖到所有可测的错误执行分支上。模拟 Read 或 Write 出错的情况，让执行流进入到 Decode 或 Encode 方法的错误分支中。

在测试代码源文件中创建了两个类型：ReturnErrorWriter 和 ReturnErrorReader，它们分别实现了 io.Writer 与 io.Reader。可以控制在第几次调用这两个类型的 Write 或 Read 方法时，返回错误，这样就可以让 Encode 或 Decode 方法按照意图，进入到不同错误分支中去。

有了这两个用例，frame 包的测试覆盖率（通过 go test -cover . 可以查看）就可以达到 90% 以上了。

4.6 Packet 的实现

和 Frame 不同，Packet 有多种类型（这里只定义了 Conn、submit、connack、submit ack)。

首先抽象这些类型需要遵循的共同接口：

// tcp-server-demo1/packet/packet.go

type Packet interface {
    Decode([]byte) error     // []byte -> struct
    Encode() ([]byte, error) //  struct -> []byte
}

Decode 是将一段字节流数据解码为一个 Packet 类型，可能是 conn，可能是 submit 等，具体要根据解码出来的 commandID 判断。

Encode 则是将一个 Packet 类型编码为一段字节流数据。

这里只完成 submit 和 submitack 类型的 Packet 接口实现，省略了 conn 流程，也省略 conn 以及 connack 类型的实现，

// tcp-server-demo1/packet/packet.go

type Submit struct {
    ID      string
    Payload []byte
}

func (s *Submit) Decode(pktBody []byte) error {
    s.ID = string(pktBody[:8])
    s.Payload = pktBody[8:]
    return nil
}

func (s *Submit) Encode() ([]byte, error) {
    return bytes.Join([][]byte{[]byte(s.ID[:8]), s.Payload}, nil), nil
}

type SubmitAck struct {
    ID     string
    Result uint8
}

func (s *SubmitAck) Decode(pktBody []byte) error {
    s.ID = string(pktBody[0:8])
    s.Result = uint8(pktBody[8])
    return nil
}

func (s *SubmitAck) Encode() ([]byte, error) {
    return bytes.Join([][]byte{[]byte(s.ID[:8]), []byte{s.Result}}, nil), nil
}

这里各种类型的编解码被调用的前提，是明确数据流是什么类型的，因此需要在包级提供一个导出的函数 Decode，这个函数负责从字节流中解析出对应的类型（根据 commandID），并调用对应类型的 Decode 方法：

// tcp-server-demo1/packet/packet.go

func Decode(packet []byte) (Packet, error) {
  commandID := packet[0]
  pktBody := packet[1:]

  switch commandID {
  case CommandConn:
    return nil, nil
  case CommandConnAck:
    return nil, nil
  case CommandSubmit:
    s := Submit{}
    err := s.Decode(pktBody)
    if err != nil {
      return nil, err
    }
    return &s, nil
  case CommandSubmitAck:
    s := SubmitAck{}
    err := s.Decode(pktBody)
    if err != nil {
      return nil, err
    }
    return &s, nil
  default:
    return nil, fmt.Errorf("unknown commandID [%d]", commandID)
  }
}

同样，也需要包级的 Encode 函数，根据传入的 packet 类型调用对应的 Encode 方法实现对象的编码：

// tcp-server-demo1/packet/packet.go

func Encode(p Packet) ([]byte, error) {
  var commandID uint8
  var pktBody []byte
  var err error

  switch t := p.(type) {
  case *Submit:
    commandID = CommandSubmit
    pktBody, err = p.Encode()
    if err != nil {
      return nil, err
    }
  case *SubmitAck:
    commandID = CommandSubmitAck
    pktBody, err = p.Encode()
    if err != nil {
      return nil, err
    }
  default:
    return nil, fmt.Errorf("unknown type [%s]", t)
  }
  return bytes.Join([][]byte{[]byte{commandID}, pktBody}, nil), nil
}

对 packet 包中各个类型的 Encode 和 Decode 方法的测试，与 frame 包的相似。

4.7 服务端的组装

服务端会将 tcp conn 与 Frame、Packet 连接起来。

第一版服务端实现：

以典型 Go 网络服务端程序的结构为基础，将 Frame、Packet 加进来

// tcp-server-demo1/cmd/server/main.go

package main

import (
  "fmt"
  "net"

  "github.com/bigwhite/tcp-server-demo1/frame"
  "github.com/bigwhite/tcp-server-demo1/packet"
)

func handlePacket(framePayload []byte) (ackFramePayload []byte, err error) {
  var p packet.Packet
  p, err = packet.Decode(framePayload)
  if err != nil {
    fmt.Println("handleConn: packet decode error:", err)
    return
  }

  switch p.(type) {
  case *packet.Submit:
    submit := p.(*packet.Submit)
    fmt.Printf("recv submit: id = %s, payload=%s\n", submit.ID, string(submit.Payload))
    submitAck := &packet.SubmitAck{
      ID:     submit.ID,
      Result: 0,
    }
    ackFramePayload, err = packet.Encode(submitAck)
    if err != nil {
      fmt.Println("handleConn: packet encode error:", err)
      return nil, err
    }
    return ackFramePayload, nil
  default:
    return nil, fmt.Errorf("unknown packet type")
  }
}

func handleConn(c net.Conn) {
  defer c.Close()
  frameCodec := frame.NewMyFrameCodec()

  for {
    // decode the frame to get the payload
    framePayload, err := frameCodec.Decode(c)
    if err != nil {
      fmt.Println("handleConn: frame decode error:", err)
      return
    }

    // do something with the packet
    ackFramePayload, err := handlePacket(framePayload)
    if err != nil {
      fmt.Println("handleConn: handle packet error:", err)
      return
    }

    // write ack frame to the connection
    err = frameCodec.Encode(c, ackFramePayload)
    if err != nil {
      fmt.Println("handleConn: frame encode error:", err)
      return
    }
  }
}

func main() {
  l, err := net.Listen("tcp", ":8888")
  if err != nil {
    fmt.Println("listen error:", err)
    return
  }

  for {
    c, err := l.Accept()
    if err != nil {
      fmt.Println("accept error:", err)
      break
    }
    // start a new goroutine to handle the new connection.
    go handleConn(c)
  }
}

程序逻辑非常清晰，服务端程序监听 8888 端口，并在每次调用 Accept 方法后得到一个新连接，服务端程序将这个新连接交到一个新的 Goroutine 中处理。

新 Goroutine 的主函数为 handleConn，有了 Packet 和 Frame 这两个抽象的加持，这个函数同样拥有清晰的代码调用结构：

// handleConn的调用结构

read frame from conn
 ->frame decode
   -> handle packet
     -> packet decode
     -> packet(ack) encode
 ->frame(ack) encode
write ack frame to conn

到这里，一个基于 TCP 的自定义应用层协议的经典阻塞式的服务端就完成了。

不过这个服务端是一个简化的实现，比如没有考虑支持优雅退出、没有捕捉某个链接上出现的可能导致整个程序退出的 panic 等。

4.8 验证测试

要验证服务端的实现是否可以正常工作，需要实现一个自定义应用层协议的客户端。同样基于 frame、packet 两个包，实现了一个自定义应用层协议的客户端。

客户端 main 函数：

// tcp-server-demo1/cmd/client/main.go
func main() {
    var wg sync.WaitGroup
    var num int = 5

    wg.Add(5)

    for i := 0; i < num; i++ {
        go func(i int) {
            defer wg.Done()
            startClient(i)
        }(i + 1)
    }
    wg.Wait()
}

客户端启动了 5 个 Goroutine，模拟 5 个并发连接。

startClient 函数是每个连接的主处理函数：

func startClient(i int) {
    quit := make(chan struct{})
    done := make(chan struct{})
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        fmt.Println("dial error:", err)
        return
    }
    defer conn.Close()
    fmt.Printf("[client %d]: dial ok", i)

    // 生成payload
    rng, err := codename.DefaultRNG()
    if err != nil {
        panic(err)
    }

    frameCodec := frame.NewMyFrameCodec()
    var counter int

    go func() {
        // handle ack
        for {
            select {
            case <-quit:
                done <- struct{}{}
                return
            default:
            }

            conn.SetReadDeadline(time.Now().Add(time.Second * 1))
            ackFramePayLoad, err := frameCodec.Decode(conn)
            if err != nil {
                if e, ok := err.(net.Error); ok {
                    if e.Timeout() {
                        continue
                    }
                }
                panic(err)
            }

            p, err := packet.Decode(ackFramePayLoad)
            submitAck, ok := p.(*packet.SubmitAck)
            if !ok {
                panic("not submitack")
            }
            fmt.Printf("[client %d]: the result of submit ack[%s] is %d\n", i, submitAck.ID, submitAck.Result)
        }
    }()

    for {
        // send submit
        counter++
        id := fmt.Sprintf("%08d", counter) // 8 byte string
        payload := codename.Generate(rng, 4)
        s := &packet.Submit{
            ID:      id,
            Payload: []byte(payload),
        }

        framePayload, err := packet.Encode(s)
        if err != nil {
            panic(err)
        }

        fmt.Printf("[client %d]: send submit id = %s, payload=%s, frame length = %d\n",
            i, s.ID, s.Payload, len(framePayload)+4)

        err = frameCodec.Encode(conn, framePayload)
        if err != nil {
            panic(err)
        }

        time.Sleep(1 * time.Second)
        if counter >= 10 {
            quit <- struct{}{}
            <-done
            fmt.Printf("[client %d]: exit ok", i)
            return
        }
    }
}

关于 startClient 函数：

• 首先，startClient 函数启动了两个 Goroutine，一个负责向服务端发送 submit 消息请求，另外一个 Goroutine 则负责读取服务端返回的响应；
• 其次，客户端发送的 submit 请求的负载（payload）是由第三方包 github.com/lucasepe/codename 负责生成的，这个包会生成一些对人类可读的随机字符串，比如：firm-iron、 moving-colleen、game-nova 这样的字符串；
• 另外，负责读取服务端返回响应的 Goroutine，使用 SetReadDeadline 方法设置了读超时，这主要是考虑该 Goroutine 可以在收到退出通知时，能及时从 Read 阻塞中跳出来。

构建和运行客户端程序。

使用 make 或者 go build 命令构建：

$make
go build github.com/bigwhite/tcp-server-demo1/cmd/server
go build github.com/bigwhite/tcp-server-demo1/cmd/client

构建成功后，先启动 server 程序：

$./server
server start ok(on *.8888)

然后启动 client 程序，启动后 client 程序便会向服务端建立 5 条连接，并发送 submit 请求，client 端的部分日志如下：

$./client
[client 5]: dial ok
[client 1]: dial ok
[client 5]: send submit id = 00000001, payload=credible-deathstrike-33e1, frame length = 38
[client 3]: dial ok
[client 1]: send submit id = 00000001, payload=helped-lester-8f15, frame length = 31
[client 4]: dial ok
[client 4]: send submit id = 00000001, payload=strong-timeslip-07fa, frame length = 33
[client 3]: send submit id = 00000001, payload=wondrous-expediter-136e, frame length = 36
[client 5]: the result of submit ack[00000001] is 0
[client 1]: the result of submit ack[00000001] is 0
[client 3]: the result of submit ack[00000001] is 0
[client 2]: dial ok
... ...
[client 3]: send submit id = 00000010, payload=bright-monster-badoon-5719, frame length = 39
[client 4]: send submit id = 00000010, payload=crucial-wallop-ec2d, frame length = 32
[client 2]: send submit id = 00000010, payload=pro-caliban-c803, frame length = 29
[client 1]: send submit id = 00000010, payload=legible-shredder-3d81, frame length = 34
[client 5]: send submit id = 00000010, payload=settled-iron-monger-bf78, frame length = 37
[client 3]: the result of submit ack[00000010] is 0
[client 4]: the result of submit ack[00000010] is 0
[client 1]: the result of submit ack[00000010] is 0
[client 2]: the result of submit ack[00000010] is 0
[client 5]: the result of submit ack[00000010] is 0
[client 4]: exit ok
[client 1]: exit ok
[client 3]: exit ok
[client 5]: exit ok
[client 2]: exit ok

client 在每条连接上发送 10 个 submit 请求后退出。

这期间服务端会输出如下日志：

recv submit: id = 00000001, payload=credible-deathstrike-33e1
recv submit: id = 00000001, payload=helped-lester-8f15
recv submit: id = 00000001, payload=wondrous-expediter-136e
recv submit: id = 00000001, payload=strong-timeslip-07fa
recv submit: id = 00000001, payload=delicate-leatherneck-4b12
recv submit: id = 00000002, payload=certain-deadpool-779d
recv submit: id = 00000002, payload=clever-vapor-25ce
recv submit: id = 00000002, payload=causal-guardian-4f84
recv submit: id = 00000002, payload=noted-tombstone-1b3e
... ...
recv submit: id = 00000010, payload=settled-iron-monger-bf78
recv submit: id = 00000010, payload=pro-caliban-c803
recv submit: id = 00000010, payload=legible-shredder-3d81
handleConn: frame decode error: EOF
handleConn: frame decode error: EOF
handleConn: frame decode error: EOF
handleConn: frame decode error: EOF
handleConn: frame decode error: EOF

从结果来看，运行正常！

项目源码：https://github.com/bigwhite/publication/tree/master/column/timegeek/go-first-course/37