3.2 长度前缀、粘包与拆包
上一节我们定好了固定头：length(4) | cmd(2) | seq(4) | flags(1)。但还有一个更靠前的问题没解决——接收方怎么知道「从哪个字节开始是一条新消息」？
TCP 不回答这个问题。它只保证字节按顺序到达，不保证「你 send 一次，对端就 read 一次」：你连续 write 的三条消息，对端可能一次 read 全收到（粘包），也可能只收到半条（拆包）。这一节要把这条没有边界的字节流还原成一条条边界清晰的消息——它是全卷最基础的一节，后面所有章节拿到的都是这里切好的 Frame，切错了后面全错。
粘包与拆包是怎么发生的
先看现象。客户端依次发送三条消息 A、B、C：
客户端三次 write 服务端一次 read 可能收到
[帧A][帧B][帧C] ───► [帧A|帧B|帧C前半] ← 粘包：三条挤在一次
[帧C后半|帧D开头] ← 拆包：一条被劈开
两种现象同源：TCP 只承诺「有序的字节」，不承诺「按 write 分组的字节」。内核的发送缓冲会把相邻的小包合并（Nagle 算法），接收缓冲则按自己方便的大小交给应用层——所以应用层绝不能假设「一次 read = 一条消息」。处理办法只有一条：给字节流加边界，最简单也最常用的边界就是长度前缀，在每条消息前面写死「这条消息有多少字节」。
长度前缀：把流切成帧
aegis 的 wire 格式如下，固定头 11 字节，length 统计它自身之后的字节数：
offset: 0 4 6 10 11 11+N
+--------+--------+--------+--------+------------------+
| length | cmd | seq | flags | payload |
| 4B BE | 2B BE | 4B BE | 1B | N B |
+--------+--------+--------+--------+------------------+
length = 7 + N（7 = cmd(2)+seq(4)+flags(1)），帧总长 = 11 + N
接收方的切帧算法只有三步：读满 11 字节固定头 → 从 length 算出载荷长度 N → 再读满 N 字节载荷。length 不含自身 4 字节，是为了让接收方读到头就能立刻知道「后面还差多少」。为什么不用分隔符（比如换行）？分隔符要求载荷里不能出现分隔符，要么转义、要么扫描，而二进制载荷里出现任意字节是常态，扫描代价高且容易出错；长度前缀直接给出长度，一次到位。
长度上限：不可信的长度字段
长度前缀有个致命前提：length 来自网络，是不可信输入。如果服务端老老实实按 length 分配缓冲，一个只发 11 字节、却声明 length = 0xFFFFFF00（约 4 GiB）的畸形包，就能让服务端为一条消息预分配 4 GiB——几百个这样的包就能把内存打满，这是最省事的一种 DoS。防线是 MaxPayload：解码时先看 length 推出的载荷长度，一旦超过上限，在分配任何内存之前就返回 ErrPayloadTooLarge 并断开连接。上限取多大是业务问题——16 MiB 对普通消息绰绰有余，真正的大消息（比如批量同步）应该走分片，而不是抬高上限。
同理，length 如果小于它必须覆盖的固定头（7 字节），说明这个包本身就是畸形的，返回 ErrShortFrame，同样在分配内存之前拒绝。
可运行产物：编解码与逐字节到达
下面 internal/proto/frame.go 是完整实现，层级 L1，命令 go test ./internal/proto/。它提供五个入口：Encode（一次分配）、EncodeInto（追加写，复用缓冲）、DecodeFrom（从已有字节切片切一帧）、ReadFrame（从 io.Reader 读一帧）、Decoder（bufio 包装，减少系统调用）。
// file: internal/proto/frame.go
// tier: L1
// verify: go test ./internal/proto/
package proto
import (
"bufio"
"encoding/binary"
"errors"
"io"
)
// ErrPayloadTooLarge 表示声明或请求的载荷长度超过 MaxPayload。
var ErrPayloadTooLarge = errors.New("proto: payload exceeds MaxPayload")
// Frame 是一条已经切干净的完整消息。
type Frame struct {
Cmd Cmd
Seq uint32
Flags Flags
Payload []byte
}
// Encode 把一帧编码成新分配的字节切片。适合一次性发送。
func Encode(f Frame) []byte {
buf := make([]byte, 0, HeaderSize+len(f.Payload))
return EncodeInto(buf, f)
}
// EncodeInto 把一帧追加到 dst 末尾并返回扩展后的切片。
//
// 复用同一个 dst 可以避免每帧一次堆分配：只要容量够，grow 就不重新分配。
// 这是发送路径上最容易省下的一笔开销——每帧一次 make 在高频广播下很可观。
func EncodeInto(dst []byte, f Frame) []byte {
need := HeaderSize + len(f.Payload)
start := len(dst)
dst = grow(dst, need)
binary.BigEndian.PutUint32(dst[start:start+4], uint32(HeaderSize-4+len(f.Payload)))
binary.BigEndian.PutUint16(dst[start+4:start+6], uint16(f.Cmd))
binary.BigEndian.PutUint32(dst[start+6:start+10], f.Seq)
dst[start+10] = byte(f.Flags)
copy(dst[start+HeaderSize:], f.Payload)
return dst
}
// grow 在 dst 末尾预留 n 字节并返回加长后的切片；容量足够时不重新分配。
func grow(dst []byte, n int) []byte {
if cap(dst)-len(dst) >= n {
return dst[:len(dst)+n]
}
grown := make([]byte, len(dst)+n)
copy(grown, dst)
return grown
}
// DecodeFrom 从 buf 头部解出一帧，返回帧与消耗的字节数。
//
// 数据不足一整帧时返回 io.ErrUnexpectedEOF，且消耗字节数为 0——调用方应保留
// 未消费的 buf 等下一次读取补齐，而不是丢弃它。
func DecodeFrom(buf []byte) (Frame, int, error) {
h, err := DecodeHeader(buf)
if err != nil {
return Frame{}, 0, err // buf 不足一个固定头 → ErrShortFrame
}
if h.Length < uint32(HeaderSize-4) {
// 长度字段比它必须覆盖的 cmd+seq+flags 还短，是畸形包。
return Frame{}, 0, ErrShortFrame
}
payloadLen := int(h.Length) - (HeaderSize - 4)
if payloadLen > MaxPayload {
return Frame{}, 0, ErrPayloadTooLarge
}
total := HeaderSize + payloadLen
if len(buf) < total {
return Frame{}, 0, io.ErrUnexpectedEOF
}
payload := make([]byte, payloadLen)
copy(payload, buf[HeaderSize:total])
return Frame{Cmd: h.Cmd, Seq: h.Seq, Flags: h.Flags, Payload: payload}, total, nil
}
// ReadFrame 从 r 读出一帧：先读满固定头，再按声明长度读满载荷。
//
// 全部读取都走 io.ReadFull，它会把「分几次到达」的字节补齐成一次完整的读；
// 这正是 TCP 流式语义下拆包的关键。读到一半连接断了会返回 io.ErrUnexpectedEOF。
func ReadFrame(r io.Reader) (Frame, error) {
var head [HeaderSize]byte
if _, err := io.ReadFull(r, head[:]); err != nil {
return Frame{}, err
}
h, err := DecodeHeader(head[:])
if err != nil {
return Frame{}, err
}
if h.Length < uint32(HeaderSize-4) {
return Frame{}, ErrShortFrame
}
payloadLen := int(h.Length) - (HeaderSize - 4)
if payloadLen > MaxPayload {
return Frame{}, ErrPayloadTooLarge
}
payload := make([]byte, payloadLen)
if _, err := io.ReadFull(r, payload); err != nil {
return Frame{}, err
}
return Frame{Cmd: h.Cmd, Seq: h.Seq, Flags: h.Flags, Payload: payload}, nil
}
// Decoder 在 bufio.Reader 上按帧解码，把大量小 read 合并成少量系统调用。
type Decoder struct {
r *bufio.Reader
}
// NewDecoder 用给定缓冲区大小包装 r（小于 16 时 bufio 会取默认值）。
func NewDecoder(r io.Reader, bufSize int) *Decoder {
return &Decoder{r: bufio.NewReaderSize(r, bufSize)}
}
// Decode 读出下一帧。流结束且正好落在帧边界时返回 io.EOF。
func (d *Decoder) Decode() (Frame, error) {
return ReadFrame(d.r)
}
测试文件把这一节最容易出错的几种到达方式全钉死：往返一致、逐字节到达（自定义 io.Reader 每次只给 1 字节，验证 io.ReadFull 会补齐）、三帧粘在一起逐帧切净、截断返回 io.ErrUnexpectedEOF、超长返回 ErrPayloadTooLarge、长度小于固定头返回 ErrShortFrame。
// file: internal/proto/frame_test.go
// tier: L1
// verify: go test ./internal/proto/
package proto
import (
"bytes"
"encoding/binary"
"io"
"testing"
)
func TestRoundTrip(t *testing.T) {
f := Frame{Cmd: CmdMoveReq, Seq: 42, Flags: FlagRequest, Payload: []byte("move 12,34")}
buf := Encode(f)
if len(buf) != HeaderSize+len(f.Payload) {
t.Fatalf("编码长度 %d", len(buf))
}
got, n, err := DecodeFrom(buf)
if err != nil || n != len(buf) {
t.Fatalf("DecodeFrom: n=%d err=%v", n, err)
}
if got.Cmd != f.Cmd || got.Seq != f.Seq || got.Flags != f.Flags || !bytes.Equal(got.Payload, f.Payload) {
t.Fatalf("往返不一致：%+v", got)
}
}
func TestEncodeIntoReusesCapacity(t *testing.T) {
dst := make([]byte, 0, 256)
base := dst
dst = EncodeInto(dst, Frame{Cmd: CmdChatSend, Seq: 1, Payload: []byte("hi")})
dst = EncodeInto(dst, Frame{Cmd: CmdChatRecv, Seq: 2, Payload: []byte("yo")})
if &dst[0] != &base[:1][0] {
t.Fatal("容量足够时 EncodeInto 不应重新分配")
}
f1, n1, _ := DecodeFrom(dst)
f2, n2, _ := DecodeFrom(dst[n1:])
if f1.Cmd != CmdChatSend || f2.Cmd != CmdChatRecv || n1+n2 != len(dst) {
t.Fatalf("切帧错误：%+v %+v", f1, f2)
}
}
// oneByteReader 每次只吐 1 字节，模拟 TCP 把包切碎后逐字节到达。
type oneByteReader struct {
data []byte
pos int
}
func (r *oneByteReader) Read(p []byte) (int, error) {
if r.pos >= len(r.data) {
return 0, io.EOF
}
if len(p) == 0 {
return 0, nil
}
p[0] = r.data[r.pos]
r.pos++
return 1, nil
}
func TestReadFrameByteByByte(t *testing.T) {
f := Frame{Cmd: CmdEnterWorldReq, Seq: 7, Flags: FlagRequest, Payload: []byte("dawn-realm")}
got, err := ReadFrame(&oneByteReader{data: Encode(f)})
if err != nil || got.Cmd != f.Cmd || !bytes.Equal(got.Payload, f.Payload) {
t.Fatalf("逐字节到达解码错误：%+v err=%v", got, err)
}
}
func TestThreeGluedFrames(t *testing.T) {
frames := []Frame{
{Cmd: CmdHeartbeatReq, Seq: 1},
{Cmd: CmdMoveReq, Seq: 2, Flags: FlagRequest, Payload: []byte("aaaaaaaa")},
{Cmd: CmdChatRecv, Seq: 3, Flags: FlagPush, Payload: []byte("hello world!")},
}
var stream []byte
for _, f := range frames {
stream = EncodeInto(stream, f)
}
off := 0
for i, want := range frames {
got, n, err := DecodeFrom(stream[off:])
if err != nil || got.Cmd != want.Cmd || got.Seq != want.Seq || !bytes.Equal(got.Payload, want.Payload) {
t.Fatalf("第 %d 帧：%+v err=%v", i, got, err)
}
off += n
}
if off != len(stream) {
t.Fatalf("三帧共消耗 %d，流长 %d（未切净）", off, len(stream))
}
}
func TestTruncated(t *testing.T) {
wire := Encode(Frame{Cmd: CmdMoveReq, Seq: 9, Payload: []byte("abcdef")})
if _, _, err := DecodeFrom(wire[:len(wire)-3]); err != io.ErrUnexpectedEOF {
t.Fatalf("截断载荷应返回 io.ErrUnexpectedEOF，得到 %v", err)
}
if _, _, err := DecodeFrom(wire[:5]); err != ErrShortFrame {
t.Fatalf("截断头应返回 ErrShortFrame，得到 %v", err)
}
if _, err := ReadFrame(bytes.NewReader(wire[:len(wire)-1])); err != io.ErrUnexpectedEOF {
t.Fatalf("ReadFrame 截断应返回 io.ErrUnexpectedEOF，得到 %v", err)
}
}
func TestPayloadTooLarge(t *testing.T) {
head := make([]byte, HeaderSize)
binary.BigEndian.PutUint32(head[0:4], uint32(HeaderSize-4+MaxPayload+1))
binary.BigEndian.PutUint16(head[4:6], uint16(CmdMoveReq))
if _, _, err := DecodeFrom(head); err != ErrPayloadTooLarge {
t.Fatalf("超长载荷应返回 ErrPayloadTooLarge，得到 %v", err)
}
if _, err := ReadFrame(bytes.NewReader(head)); err != ErrPayloadTooLarge {
t.Fatalf("ReadFrame 超长应返回 ErrPayloadTooLarge，得到 %v", err)
}
}
func TestLengthBelowHeader(t *testing.T) {
head := make([]byte, HeaderSize)
binary.BigEndian.PutUint32(head[0:4], 6) // 小于 length 之后的固定头 7 字节
if _, _, err := DecodeFrom(head); err != ErrShortFrame {
t.Fatalf("长度小于固定头应返回 ErrShortFrame，得到 %v", err)
}
}
func TestDecoderEOF(t *testing.T) {
var stream []byte
stream = EncodeInto(stream, Frame{Cmd: CmdLoginReq, Seq: 1, Payload: []byte("v=3")})
stream = EncodeInto(stream, Frame{Cmd: CmdLoginResp, Seq: 1, Flags: FlagResponse, Payload: []byte("ok")})
d := NewDecoder(bytes.NewReader(stream), 64)
for i := 0; i < 2; i++ {
if _, err := d.Decode(); err != nil {
t.Fatalf("第 %d 帧：%v", i, err)
}
}
if _, err := d.Decode(); err != io.EOF {
t.Fatalf("流末尾应返回 io.EOF，得到 %v", err)
}
}
本机实跑：ok aegis/internal/proto，本节这 8 条全绿。最有价值的是 TestReadFrameByteByByte——它把每个字节单独喂进去，等于把「拆包」推到极端：如果解码器里有任何一处假设「一次 read 拿到一整帧」，这条测试立刻失败。它之所以能过，是因为 ReadFrame 全程用 io.ReadFull。
本节小结
- TCP 只承诺有序字节，不承诺 write 边界；粘包与拆包同源，处理办法是给流加边界。
- 长度前缀
length(4, BE) | cmd(2) | seq(4) | flags(1) | payload(N)，length = 7 + N，切帧只需「读头 → 算长 → 读满」三步。 - 长度字段不可信：
MaxPayload必须在分配内存之前拦下超长声明，否则畸形包就能打满内存。 EncodeInto复用缓冲、io.ReadFull补齐字节，是发送与接收两侧最关键的两行。
下一节处理协议演进：客户端版本落后或超前时，服务端怎么协商、怎么兼容。
阅读导航：上一节：3.1 消息头与命令号 · 下一节：3.3 协议版本与多版本兼容 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。