3、数据包捕获：使用Python接收原始数据流、数据缓冲与粘包处理

好，咱们进入实战环节。前面讲了行情数据的格式，也讲了怎么从网卡抓包。但真正把数据从网络里「捞」出来，送到你的策略里，中间还有几个坑要填。

说白了，网络不保证你每次 recv() 拿到的都是一个完整的行情包。它可能给你半个包，也可能一次塞给你三个包。这就是所谓的粘包和拆包问题。

我刚开始做量化那会儿，就吃过这个亏。记得有一次回测跑得挺漂亮，一上实盘就报错。查了半天，发现是数据包没对齐，解析出来的价格全是乱码。嗯，从那以后，我再也不敢小看数据缓冲这步了。

3.1 原始数据流的接收方式

Python 里接收网络数据，最常用的就是 socket 模块。对于行情这种低延迟场景，我建议直接用 TCP 或 UDP 的原始 socket。

先看一个最简单的 TCP 接收示例：

import socket

def create_tcp_connection(host, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    sock.connect((host, port))
    return sock

sock = create_tcp_connection('192.168.1.100', 8888)
buffer = b''
while True:
    chunk = sock.recv(4096)
    if not chunk:
        break
    buffer += chunk
    # 后续处理...

这里有个细节：TCP_NODELAY 这个选项。我个人的习惯是，做行情接收时一定要开。它禁用了 Nagle 算法，避免因为小包合并而增加延迟。你想想看，行情数据本来就讲究毫秒级，再让系统帮你攒一攒再发，黄花菜都凉了。

3.2 数据缓冲区的设计

为什么需要缓冲区？因为网络数据是流式的。你调用 recv() 时，底层可能只到了 200 字节，但一个完整的行情包可能是 300 字节。这时候你就得先把这 200 字节存起来，等剩下的 100 字节到了再拼起来处理。

我常用的缓冲区设计有两种：

方案 优点 缺点 适用场景
字节串拼接 实现简单，代码直观 频繁拼接产生大量临时对象 低频行情，每秒几百笔
bytearray 预分配 内存复用，GC压力小 需要管理读写指针 高频行情，每秒上万笔

对于大多数场景，字节串拼接就够了。但如果你做的是 Level-2 行情，每秒几万笔数据进来，我建议用 bytearray。我曾经在一个项目中，就因为频繁的 bytes 拼接导致 GC 停顿，行情延迟从 50 微秒飙到了 2 毫秒。嗯，这个教训挺深刻的。

来看一个 bytearray 缓冲区的实现：

class RingBuffer:
    def __init__(self, size=65536):
        self.buffer = bytearray(size)
        self.read_pos = 0
        self.write_pos = 0
        self.size = size

    def write(self, data):
        length = len(data)
        # 检查剩余空间
        available = self.size - (self.write_pos - self.read_pos)
        if length > available:
            raise BufferError('缓冲区满了，需要扩容')
        # 写入数据
        end = self.write_pos + length
        if end <= self.size:
            self.buffer[self.write_pos:end] = data
        else:
            # 环形写入
            first_part = self.size - self.write_pos
            self.buffer[self.write_pos:] = data[:first_part]
            self.buffer[:length - first_part] = data[first_part:]
        self.write_pos = end % self.size

    def read(self, length):
        if length > self.available():
            return None
        if self.read_pos + length <= self.size:
            data = bytes(self.buffer[self.read_pos:self.read_pos + length])
        else:
            first_part = self.size - self.read_pos
            data = bytes(self.buffer[self.read_pos:] + 
                         self.buffer[:length - first_part])
        self.read_pos = (self.read_pos + length) % self.size
        return data

    def available(self):
        return self.write_pos - self.read_pos
小提示：环形缓冲区的大小要设置合理。太小了容易丢包，太大了浪费内存。我一般按「峰值带宽 × 最大延迟」来估算。比如峰值 100MB/s，网络延迟 10ms，那缓冲区至少需要 1MB。

3.3 粘包处理的核心逻辑

粘包，说白了就是你一次 recv() 拿到的数据里包含了多个完整的包，或者一个包被拆成了多次 recv()。怎么解决？核心思路就四个字：定界分包。

常见的定界方式有三种：

  1. 固定长度：每个包都是 100 字节，直接按长度切分。简单粗暴，但浪费带宽。
  2. 长度前缀：包头里写明了包体长度。这是最常用的方式。
  3. 特殊分隔符：比如用 \r\n 结尾。适合文本协议，二进制协议不推荐。

行情协议绝大多数用的是第二种——长度前缀。比如深交所的行情，前 4 个字节就是包体长度。

来看一个通用的粘包处理函数：

def parse_packets(buffer, header_size=4):
    """
    从缓冲区中解析出完整的包
    返回: (完整包列表, 剩余未处理数据)
    """
    packets = []
    offset = 0
    buffer_len = len(buffer)

    while offset + header_size <= buffer_len:
        # 读取包头中的长度字段
        packet_length = int.from_bytes(
            buffer[offset:offset + header_size], 
            byteorder='little'
        )
        # 检查是否收到了完整的包
        if offset + header_size + packet_length > buffer_len:
            break  # 包还没收全，等下一次 recv()
        # 提取完整包
        full_packet = buffer[offset:offset + header_size + packet_length]
        packets.append(full_packet)
        offset += header_size + packet_length

    # 剩余未处理的数据
    remaining = buffer[offset:]
    return packets, remaining
注意：长度字段的字节序一定要和协议一致。我曾经帮一个同事排查问题，他死活解析不对，最后发现是 byteorder 写成了 'big'，而协议用的是 'little'。就这一个字节序的问题，折腾了两天。

3.4 完整的数据接收循环

把上面这些串起来，就是一个完整的数据接收循环：

import socket
import struct

class MarketDataReceiver:
    def __init__(self, host, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.sock.connect((host, port))
        self.buffer = b''

    def recv_loop(self):
        while True:
            try:
                chunk = self.sock.recv(65536)
                if not chunk:
                    print('连接已断开')
                    break
                self.buffer += chunk
                packets, self.buffer = parse_packets(self.buffer)
                for packet in packets:
                    self.handle_packet(packet)
            except socket.timeout:
                continue
            except Exception as e:
                print(f'接收异常: {e}')
                break

    def handle_packet(self, packet):
        # 这里交给具体的解码模块处理
        # 比如解析出 股票代码、最新价、成交量 等
        pass

这个循环看起来简单，但有几个地方要注意：

  • recv 的超时设置：不要无限阻塞，设个 100ms 的超时，方便做心跳检测。
  • 异常处理：网络环境复杂，断线重连是常态。我一般会加个重试机制，最多重试 3 次，每次间隔 1 秒。
  • 缓冲区清理：如果某个包一直没收全，缓冲区会越来越大。我建议设个最大长度，超过就丢弃。

3.5 核心逻辑流程图

下面这张图，把整个数据接收、缓冲、粘包处理的流程串起来了。你可以对照着代码看，会更清楚。

数据包捕获与粘包处理流程 1. socket.recv() 2. buffer += chunk 3. parse_packets() 完整? 4. 解码处理 handle_packet() 等待更多数据 继续 recv() 关键点： • 每次 recv() 后立即拼入缓冲区，不要等 • 解析时只取完整包，剩余数据留在缓冲区

3.6 避坑指南

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

  • 不要假设 recv() 一次就能收全。TCP 是流协议，它不保证一次 recv() 就是一个完整的应用层包。我见过太多新手直接 recv(100) 然后当完整包处理，结果数据全是乱的。
  • 小心内存泄漏。如果某个包一直没收全，缓冲区会无限增长。我一般设个上限，比如 10MB，超过就丢弃并重新连接。
  • 多线程接收要加锁。如果你用多线程同时 recv()，缓冲区操作必须加锁。不过我个人建议用单线程 + 异步 IO，省心很多。
核心总结：数据接收的核心就三件事——收数据、拼缓冲区、按长度切包。把这三点做好，粘包问题就解决了八成。剩下的两成，靠异常处理和重连机制兜底。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →