第三章 固定长度协议：最朴实的拆包方案

说实话，刚入行那会儿，我遇到粘包问题第一反应就是「加个特殊分隔符呗」。后来在做一个工业数据采集项目时，发现数据里偶尔会出现和分隔符一模一样的字节，结果整个解析逻辑就崩了。那次教训让我明白——在某些场景下，固定长度协议反而是最稳的选择。

固定长度协议，说白了就是通信双方约定好：每个消息包的长度是固定的。比如约定 1024 字节，那发送方每次就发 1024 字节，接收方每次也读 1024 字节。不够？补零。多了？截断。简单粗暴，但有效。

核心思想：用空间换确定性。既然无法预知网络什么时候会粘包，那就让每个包都一样长，接收方按固定尺寸切分即可。

3.1 固定长度协议的适用场景

不是所有场景都适合用固定长度。我个人习惯在以下几种情况优先考虑它：

  • 数据字段长度稳定：比如传感器读数、GPS 坐标、状态心跳包，每个字段的字节数基本不变
  • 实时性要求高：不需要扫描分隔符，直接按长度切，解析速度极快
  • 嵌入式或资源受限设备：MCU 上做字符串匹配分隔符很费劲，固定长度用 memcpy 就能搞定
  • 协议设计初期：团队还没想好复杂协议时，先用固定长度快速验证功能

但也要注意它的短板。我在一个视频传输项目里试过固定长度——视频帧大小差异很大，补零浪费了 60% 的带宽，后来还是换成了变长方案。所以，固定长度适合「数据长度波动小」的场景。

场景 固定长度 分隔符 TLV 变长
传感器数据（每包 64 字节） ✅ 推荐 ⚠️ 可用 ⚠️ 可用
视频帧（大小 1KB~1MB） ❌ 浪费带宽 ⚠️ 需转义 ✅ 推荐
心跳包（固定 8 字节） ✅ 极简 ✅ 可用 ⚠️ 冗余
文本日志（长度不定） ❌ 补零太多 ✅ 推荐 ✅ 推荐

3.2 固定长度拆包的核心逻辑

你想想看，接收方其实就做一件事：从缓冲区里按固定长度切数据。如果缓冲区不够一个完整包，就等下一次接收。流程图如下：

固定长度拆包流程 接收原始数据，追加到缓冲区 缓冲区长度 ≥ 固定包长？ 是：切出前 N 字节 否：等待更多数据 处理完整包 缓冲区前移，继续检查 ⚠️ 注意 切出完整包后，缓冲区剩余数据 可能还包含下一个完整包， 需要循环检查。

3.3 Python 实现：固定长度拆包器

下面是我在实际项目中用过的拆包器代码。我习惯把它封装成一个类，方便复用。

import struct

class FixedLengthUnpacker:
    """
    固定长度拆包器
    每次接收数据后调用 feed()，通过回调返回完整包
    """
    def __init__(self, packet_size: int, callback=None):
        if packet_size <= 0:
            raise ValueError("packet_size 必须大于 0")
        self.packet_size = packet_size
        self.buffer = b""
        self.callback = callback

    def feed(self, data: bytes):
        """喂入新收到的数据"""
        self.buffer += data
        # 循环拆包：缓冲区可能包含多个完整包
        while len(self.buffer) >= self.packet_size:
            packet = self.buffer[:self.packet_size]
            self.buffer = self.buffer[self.packet_size:]
            if self.callback:
                self.callback(packet)

    def reset(self):
        """重置缓冲区（比如连接断开时）"""
        self.buffer = b""

# 使用示例
def on_packet(pkt):
    print(f"收到完整包: {pkt.hex()}")

unpacker = FixedLengthUnpacker(packet_size=16, callback=on_packet)

# 模拟粘包：一次发送两个包
unpacker.feed(b"AAAAAAAABBBBBBBB")  # 两个 8 字节包粘在一起
# 输出：
# 收到完整包: 4141414141414141
# 收到完整包: 4242424242424242

我的习惯：固定包长建议定义为类的常量，不要硬编码。如果协议升级需要改包长，改一处就行。另外，回调函数里不要做耗时操作——拆包器只负责拆包，业务逻辑交给上层。

3.4 处理半包：缓冲区不够一个完整包

网络数据不会每次都恰好对齐包边界。比如包长 1024 字节，但这次只收到 500 字节。怎么办？

答案很简单：等。把 500 字节留在缓冲区，下次 recv 再收到数据时追加进去，直到缓冲区长度 ≥ 1024，再切包。

上面的代码已经处理了这种情况——while 循环条件保证了只有缓冲区够长才切包。不够？直接跳过，等下次 feed()。

我曾经踩过的坑：在嵌入式设备上，缓冲区大小是有限的。如果对方发了一个超大包（比如 64KB），而你的缓冲区只有 8KB，就会一直收不齐。建议给缓冲区设置上限，超过上限就丢弃或重置，防止内存耗尽。

3.5 固定长度协议的变体：固定头 + 固定体

有些场景下，整个包固定长度太浪费。我做过一个交易系统，消息头固定 8 字节（含消息类型和长度），消息体长度可变但不超过 1024 字节。这种其实算「固定头 + 定长体」的混合方案。

但注意，它本质上还是固定长度——因为消息体虽然可变，但协议约定「消息体最大 1024 字节，不足补零」。接收方先读 8 字节头，再读 1024 字节体，总共 1032 字节一个包。

class FixedHeaderBodyUnpacker:
    """
    固定头(8字节) + 固定体(1024字节) 拆包
    头结构: 2字节消息类型 + 2字节保留 + 4字节序列号
    """
    HEADER_SIZE = 8
    BODY_SIZE = 1024
    PACKET_SIZE = HEADER_SIZE + BODY_SIZE

    def __init__(self, callback=None):
        self.buffer = b""
        self.callback = callback

    def feed(self, data: bytes):
        self.buffer += data
        while len(self.buffer) >= self.PACKET_SIZE:
            packet = self.buffer[:self.PACKET_SIZE]
            self.buffer = self.buffer[self.PACKET_SIZE:]
            if self.callback:
                self.callback(packet)

    @staticmethod
    def parse_header(packet: bytes):
        msg_type = struct.unpack("!H", packet[0:2])[0]
        seq_num = struct.unpack("!I", packet[4:8])[0]
        return msg_type, seq_num

3.6 固定长度 vs 其他方案：怎么选？

我整理了一个对比表，方便你根据项目情况做决策：

维度 固定长度 分隔符 TLV 变长
实现复杂度 ⭐ 极低 ⭐⭐ 低 ⭐⭐⭐ 中
解析速度 最快（O(1) 切分） 中（需扫描分隔符） 中（需解析长度字段）
带宽利用率 低（补零浪费） 高（仅多 1~2 字节） 高（精确控制）
数据安全性 高（不会误判） 低（数据含分隔符会出错） 高（长度字段校验）
适用场景 定长数据、嵌入式 文本协议、日志 通用、复杂协议

我的建议：如果你刚开始设计协议，或者数据字段长度基本固定，别犹豫，直接用固定长度。它虽然浪费一点带宽，但换来的是极低的 bug 率和极快的开发速度。等业务复杂了再升级到 TLV 也不迟。

3.7 避坑指南

最后分享几个我实际踩过的坑：

  • 包长定义不一致：服务端用 1024，客户端用 2048，结果两边永远对不上。一定要把包长定义放在共享的常量文件里。
  • 补零字符冲突：如果数据本身包含 \x00，补零用 \x00 会导致无法区分「有效数据」和「填充」。我习惯用 \x00 填充，但业务层需要知道实际数据长度（可以在包头加一个长度字段）。
  • 缓冲区无限增长：如果对方恶意发送大量半包，你的缓冲区会一直膨胀。建议设置最大缓冲区大小，超过就断开连接。
  • 大端小端问题：如果包头里有长度字段，记得统一用网络字节序（大端）。我在跨平台项目里吃过这个亏——x86 是小端，ARM 是大端，解析出来的长度完全不对。
← 上一章 📖 返回目录 下一章 →