golang标准库对io的抽象非常精巧，各个组件可以随意组合，可以作为接口设计的典范。这篇文章结合一个实际的例子来和大家分享一下。
背景
以一个RPC的协议包来说，每个包有如下结构
1
2
3
4
5
6
|
type Packet struct {
TotalSize uint32
Magic [4]byte
Payload []byte
Checksum uint32
}
|
其中TotalSize是整个包除去TotalSize后的字节数， Magic是一个固定长度的字串，Payload是包的实际内容，包含业务逻辑的数据。
Checksum是对Magic和Payload的adler32校验和。
编码(encode)
我们使用一个原型为func EncodePacket(w io.Writer, payload []byte) error的函数来把数据打包，结合encoding/binary我们很容易写出第一版，演示需要，错误处理方面就简化处理了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
var RPC_MAGIC = [4]byte{'p', 'y', 'x', 'i'}
func EncodePacket(w io.Writer, payload []byte) error {
// len(Magic) + len(Checksum) == 8
totalsize := uint32(len(payload) + 8)
// write total size
binary.Write(w, binary.BigEndian, totalsize)
// write magic bytes
binary.Write(w, binary.BigEndian, RPC_MAGIC)
// write payload
w.Write(payload)
// calculate checksum
var buf bytes.Buffer
buf.Write(RPC_MAGIC[:])
buf.Write(payload)
checksum := adler32.Checksum(buf.Bytes())
// write checksum
return binary.Write(w, binary.BigEndian, checksum)
}
|
在上面的实现中，为了计算checksum，我们使用了一个内存buffer来缓存数据，最后把所有的数据一次性读出来算checksum，考虑到计算checksum是一个不断update地过程，我们应该有方法直接略过内存buffer而计算checksum。
查看hash/adler32我们得知，我们可以构造一个Hash32的对象，这个对象内嵌了一个Hash的接口，这个接口的定义如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
type Hash interface {
// Write (via the embedded io.Writer interface) adds more data to the running hash.
// It never returns an error.
io.Writer
// Sum appends the current hash to b and returns the resulting slice.
// It does not change the underlying hash state.
Sum(b []byte) []byte
// Reset resets the Hash to its initial state.
Reset()
// Size returns the number of bytes Sum will return.
Size() int
// BlockSize returns the hash's underlying block size.
// The Write method must be able to accept any amount
// of data, but it may operate more efficiently if all writes
// are a multiple of the block size.
BlockSize() int
}
|
这是一个通用的计算hash的接口，标准库里面所有计算hash的对象都实现了这个接口，比如md5, crc32等。由于Hash实现了io.Writer接口，因此我们可以把所有要计算的数据像写入文件一样写入到这个对象中，最后调用Sum(nil)就可以得到最终的hash的byte数组。利用这个思路，第二版可以这样写:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
func EncodePacket2(w io.Writer, payload []byte) error {
// len(Magic) + len(Checksum) == 8
totalsize := uint32(len(RPC_MAGIC) + len(payload) + 4)
// write total size
binary.Write(w, binary.BigEndian, totalsize)
// write magic bytes
binary.Write(w, binary.BigEndian, RPC_MAGIC)
// write payload
w.Write(payload)
// calculate checksum
sum := adler32.New()
sum.Write(RPC_MAGIC[:])
sum.Write(payload)
checksum := sum.Sum32()
// write checksum
return binary.Write(w, binary.BigEndian, checksum)
}
|
注意这次的变化，前面写入TotalSize，Magic，Payload部分没有变化，在计算checksum的时候去掉了bytes.Buffer，减少了一次内存申请和拷贝。
考虑到sum和w都是io.Writer，利用神奇的io.MultiWriter，我们可以这样写
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
func EncodePacket(w io.Writer, payload []byte) error {
// len(Magic) + len(Checksum) == 8
totalsize := uint32(len(RPC_MAGIC) + len(payload) + 4)
// write total size
binary.Write(w, binary.BigEndian, totalsize)
sum := adler32.New()
ww := io.MultiWriter(sum, w)
// write magic bytes
binary.Write(ww, binary.BigEndian, RPC_MAGIC)
// write payload
ww.Write(payload)
// calculate checksum
checksum := sum.Sum32()
// write checksum
return binary.Write(w, binary.BigEndian, checksum)
}
|
注意MultiWriter的使用，我们把w和sum利用MultiWriter绑在了一起创建了一个新的Writer，向这个Writer里面写入数据就同时向w和sum里面都写入数据，这样就完成了发送数据和计算checksum的同步进行，而对于binary.Write来说没有任何区别，因为它需要的是一个实现了Write方法的对象。
解码(decode)
基于上面的思想，解码也可以把接收数据和计算checksum一起进行，完整代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
func DecodePacket(r io.Reader) ([]byte, error) {
var totalsize uint32
err := binary.Read(r, binary.BigEndian, &totalsize)
if err != nil {
return nil, errors.Annotate(err, "read total size")
}
// at least len(magic) + len(checksum)
if totalsize < 8 {
return nil, errors.Errorf("bad packet. header:%d", totalsize)
}
sum := adler32.New()
rr := io.TeeReader(r, sum)
var magic [4]byte
err = binary.Read(rr, binary.BigEndian, &magic)
if err != nil {
return nil, errors.Annotate(err, "read magic")
}
if magic != RPC_MAGIC {
return nil, errors.Errorf("bad rpc magic:%v", magic)
}
payload := make([]byte, totalsize-8)
_, err = io.ReadFull(rr, payload)
if err != nil {
return nil, errors.Annotate(err, "read payload")
}
var checksum uint32
err = binary.Read(r, binary.BigEndian, &checksum)
if err != nil {
return nil, errors.Annotate(err, "read checksum")
}
if checksum != sum.Sum32() {
return nil, errors.Errorf("checkSum error, %d(calc) %d(remote)", sum.Sum32(), checksum)
}
return payload, nil
}
|
上面代码中，我们使用了io.TeeReader，这个函数的原型为func TeeReader(r Reader, w Writer) Reader，它返回一个Reader，这个Reader是参数r的代理，读取的数据还是来自r，不过同时把读取的数据写入到w里面。
一切皆文件
unix下有一切皆文件的思想，golang把这个思想贯彻到更远，因为本质上我们对文件的抽象就是一个可读可写的一个对象，也就是实现了io.Writer和io.Reader的对象我们都可以称为文件，在上面的例子中无论是EncodePacket还是DecodePacket我们都没有假定编码后的数据是发送到socket，还是从内存读取数据解码，因此我们可以这样调用EncodePacket
1
2
|
conn, _ := net.Dial("tcp", "127.0.0.1:8000")
EncodePacket(conn, []byte("hello"))
|
把数据直接发送到socket，也可以这样
1
2
3
|
conn, _ := net.Dial("tcp", "127.0.0.1:8000")
bufconn := bufio.NewWriter(conn)
EncodePacket(bufconn, []byte("hello"))
|
对socket加上一个buffer来增加吞吐量，也可以这样
1
2
3
4
|
conn, _ := net.Dial("tcp", "127.0.0.1:8000")
zip := zlib.NewWriter(conn)
bufconn := bufio.NewWriter(conn)
EncodePacket(bufconn, []byte("hello"))
|
加上一个zip压缩，还可以利用加上crypto/aes来个AES加密…
在这个时候，文件已经不再局限于io，可以是一个内存buffer，也可以是一个计算hash的对象，甚至是一个计数器，流量限速器。golang灵活的接口机制为我们提供了无限可能。
结尾
我一直认为一个好的语言一定有一个设计良好的标准库，golang的标准库是作者们多年系统编程的沉淀，值得我们细细品味