4. 串口数据解析与粘包处理

串口通信里，最让人头疼的问题之一就是「粘包」。数据发过来，你这边读到的可能是一整段，也可能是半截，甚至好几帧粘在一起。我刚开始做嵌入式项目时，就被这个问题折腾了好几个通宵。今天咱们就把这个坎迈过去。

4.1 常见数据帧格式设计

先聊聊数据帧长什么样。说白了，就是收发双方约定好的一套「暗号」。我习惯把帧格式分成三类：

帧类型	典型结构	适用场景
分隔符帧	数据 + \r\n	AT指令、文本协议
固定长度帧	固定N字节	传感器数据、状态上报
变长帧（含长度域）	头 + 长度 + 数据 + 校验	复杂协议、文件传输

你想想看，如果只是传个温度值，用固定长度帧最省事。但要是传AT指令，那必须用分隔符，因为指令长短不一。我见过有人用固定长度去解析AT指令，结果死活对不上——嗯，协议设计一开始就要想清楚。

4.2 基于分隔符的解析

分隔符解析，说白了就是「读到某个字符就认为一帧结束」。最常见的分隔符是 \n 或 \r\n。Python里用 readline() 就能搞定。

但有个坑——如果数据里本身就包含分隔符怎么办？我曾经在一个GPS模块项目里遇到过，数据里偶尔会出现 \r\n 的二进制片段，结果程序直接崩了。后来我加了个转义机制才解决。

import serial

ser = serial.Serial('COM3', 115200, timeout=1)

def read_line(ser):
    """读取一行数据，以\n结尾"""
    buffer = b''
    while True:
        ch = ser.read(1)
        if not ch:
            break  # 超时退出
        if ch == b'\n':
            break
        buffer += ch
    return buffer

# 使用示例
while True:
    line = read_line(ser)
    if line:
        print(f"收到: {line.decode('utf-8', errors='ignore')}")

4.3 基于固定长度的解析

固定长度解析就简单多了。你预先知道每帧是N个字节，直接读N个字节就行。但这里有个关键问题：怎么知道什么时候开始读？

我常用的做法是：先找帧头（比如 0xAA 0x55），找到后再读固定长度的数据。这样即使中间有干扰数据，也能快速恢复同步。

import serial

ser = serial.Serial('COM3', 115200, timeout=1)

FRAME_HEADER = b'\xAA\x55'
FRAME_LENGTH = 8  # 固定8字节一帧

def read_fixed_frame(ser):
    """读取固定长度帧"""
    # 先找帧头
    while True:
        header = ser.read(2)
        if header == FRAME_HEADER:
            break
        if not header:
            return None  # 超时
    
    # 读取剩余数据
    data = ser.read(FRAME_LENGTH - 2)
    if len(data) != FRAME_LENGTH - 2:
        return None  # 数据不完整
    
    return FRAME_HEADER + data

# 使用示例
while True:
    frame = read_fixed_frame(ser)
    if frame:
        print(f"收到帧: {frame.hex()}")

4.4 粘包处理实战

粘包的本质是什么？就是接收端没来得及处理，数据在缓冲区里堆在一起了。你读一次，可能读到好几帧的数据。

解决思路其实就一句话：把接收和解析分开。接收只管往缓冲区里塞数据，解析再从缓冲区里按帧规则取数据。

import serial
import threading
from collections import deque

class UartParser:
    def __init__(self, port, baudrate, frame_length=8):
        self.ser = serial.Serial(port, baudrate, timeout=0.1)
        self.buffer = bytearray()
        self.frame_length = frame_length
        self.frames = deque()
        self.running = True
        
        # 启动接收线程
        self.recv_thread = threading.Thread(target=self._recv_loop)
        self.recv_thread.start()
    
    def _recv_loop(self):
        """接收线程：只管往缓冲区里塞数据"""
        while self.running:
            data = self.ser.read(64)  # 一次最多读64字节
            if data:
                self.buffer.extend(data)
                self._parse_frames()
    
    def _parse_frames(self):
        """解析线程：从缓冲区里按帧取数据"""
        while len(self.buffer) >= self.frame_length:
            # 检查帧头
            if self.buffer[0] == 0xAA and self.buffer[1] == 0x55:
                frame = self.buffer[:self.frame_length]
                self.frames.append(bytes(frame))
                self.buffer = self.buffer[self.frame_length:]
            else:
                # 帧头不对，跳过1个字节
                self.buffer.pop(0)
    
    def get_frame(self, timeout=1):
        """获取一帧数据（非阻塞）"""
        try:
            return self.frames.popleft()
        except IndexError:
            return None
    
    def close(self):
        self.running = False
        self.recv_thread.join()
        self.ser.close()

这个设计的好处是：接收线程和主线程互不干扰。你可以在主循环里做其他事，需要数据时直接 get_frame() 就行。

4.5 实战：温湿度传感器数据读取

咱们拿一个真实的温湿度传感器（比如DHT22通过串口转接）来练手。假设它的数据帧格式是：

• 帧头： 0xAA 0x55 (2字节)
• 温度： 2字节，小端模式，单位0.1°C
• 湿度： 2字节，小端模式，单位0.1%RH
• 校验： 1字节，前面5字节的累加和
• 帧尾： 0x0D 0x0A (2字节)

总长度：2 + 2 + 2 + 1 + 2 = 9字节

import serial

ser = serial.Serial('COM3', 115200, timeout=1)

def parse_sensor_frame(frame):
    """解析温湿度传感器帧"""
    if len(frame) != 9:
        return None
    
    # 校验帧头
    if frame[0] != 0xAA or frame[1] != 0x55:
        return None
    
    # 校验帧尾
    if frame[7] != 0x0D or frame[8] != 0x0A:
        return None
    
    # 校验累加和
    checksum = sum(frame[:5]) & 0xFF
    if checksum != frame[6]:
        return None
    
    # 解析温度（小端）
    temp_raw = frame[2] | (frame[3] << 8)
    temperature = temp_raw / 10.0
    
    # 解析湿度（小端）
    hum_raw = frame[4] | (frame[5] << 8)
    humidity = hum_raw / 10.0
    
    return {'temperature': temperature, 'humidity': humidity}

# 主循环
while True:
    # 找帧头
    while True:
        b = ser.read(1)
        if b == b'\xAA':
            next_b = ser.read(1)
            if next_b == b'\x55':
                break
    
    # 读取剩余7字节
    rest = ser.read(7)
    if len(rest) == 7:
        frame = b'\xAA\x55' + rest
        result = parse_sensor_frame(frame)
        if result:
            print(f"温度: {result['temperature']:.1f}°C, "
                  f"湿度: {result['humidity']:.1f}%")
        else:
            print("校验失败，丢弃该帧")

4.6 知识结构图

下面这张图帮你理清串口数据解析的整体思路：

4.7 避坑指南

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

• 超时设置太短： 传感器响应时间可能不稳定，我建议设成1秒以上。
• 忘记清缓冲区： 程序启动时最好清一次串口缓冲区，避免读到上次残留数据。
• 校验算法不一致： 收发双方的校验算法必须完全一致，包括初始值和计算范围。
• 多线程共享变量： 缓冲区操作要加锁，否则会出现数据竞争。