Python中快速提取二进制数据包起始地址的优化方案

问题背景 #

我有一个由数据包序列化而成的二进制文件，结构如下：

[Length][Payload][Length][Payload][Length][Payload]

Length字段固定为4字节，数值不固定且无规律，其值包含自身4字节的长度。需要提取文件中每个[Length]字段首字节的字节位置，示例如下：

00 00 02 00 FF FF FF FF FF ...
00 01 3C E5 FF FF FF FF FF ...
00 00 A5 90 FF FF FF FF FF ...
^
需保存所有此类索引

现有实现及性能问题 #

原mmap版本代码 #

data = mmap.mmap(filename.fileno(), 0, access=mmap.ACCESS_READ)
fileSize = os.path.getsize(filename.name)

address = 0
addresses = []

start = time.time()

while address < fileSize:
    pkt_length = bytes2int(data[address:(address + 4)])
    addresses.append(address)
    address += pkt_length

end = time.time()

print(len(addresses))
print(end-start)

性能数据（编辑1）

处理4.2GB文件时：

45559456
21.271047115325928

该版本内存占用较高，且处理超10万个数据包时速度偏慢。

os库实现版本（编辑2） #

start = time.time()
addresses = []
fp = os.open(fileName, os.O_RDONLY)
while True:
    buf = os.read(fp, 4)
    if not buf:
        break
    size = int.from_bytes(buf, byteorder='big')
    addresses.append(os.lseek(fp, size-4, os.SEEK_CUR))  # -4 cause length already read

end = time.time()
print(len(addresses))
print(end-start)

性能数据

处理同一份4.2GB文件时：

45559456
179.0517611503601

该版本速度更慢，因为频繁的系统调用（os.read和os.lseek）开销极大。

优化方案 #

1. 用紧凑数据结构存储地址 #

Python列表存储整数的内存开销大，改用array模块存储地址，它会以连续的二进制形式存储数据，大幅降低内存占用，同时append操作也更快。

示例代码：

import mmap
import os
import time
import array

def get_packet_addresses(filename):
    addresses = array.array('Q')  # 无符号长整型，适配64位地址
    with open(filename, 'rb') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as data:
            file_size = len(data)
            address = 0
            while address < file_size:
                # 使用内置int.from_bytes替代自定义bytes2int，速度更快
                pkt_length = int.from_bytes(data[address:address+4], byteorder='big')
                addresses.append(address)
                address += pkt_length
    return addresses

start = time.time()
addrs = get_packet_addresses('your_file.bin')
end = time.time()

print(len(addrs))
print(end - start)

2. 批量读取解析，减少循环次数 #

将单条解析改为批量读取一段数据，一次性解析多个Length字段，降低Python循环的开销（Python循环本身是解释执行，次数越多开销越大）。

示例代码：

import mmap
import os
import time
import array

def get_packet_addresses_batch(filename):
    addresses = array.array('Q')
    batch_size = 1024 * 1024  # 每次处理1MB数据，可根据内存调整
    with open(filename, 'rb') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as data:
            file_size = len(data)
            address = 0
            while address < file_size:
                end_batch = min(address + batch_size, file_size)
                batch_data = data[address:end_batch]
                offset = 0
                # 在当前批次内解析所有完整的Length字段
                while offset + 4 <= len(batch_data):
                    pkt_length = int.from_bytes(batch_data[offset:offset+4], byteorder='big')
                    addresses.append(address + offset)
                    offset += pkt_length
                # 更新全局地址到批次处理后的位置
                address += offset
    return addresses

start = time.time()
addrs = get_packet_addresses_batch('your_file.bin')
end = time.time()

print(len(addrs))
print(end - start)

3. 预分配内存（可选） #

如果能预估数据包的大致数量，可以提前给array或列表预分配空间，避免动态扩容的开销。比如：

estimated_count = 50000000  # 预估5000万个数据包
addresses = array.array('Q', [0]) * estimated_count
# 后续用索引赋值替代append

优化原理说明 #

• 紧凑数据结构：array.array('Q')每个元素仅占8字节，相比Python列表每个int占28字节，内存占用减少约70%，同时内存连续性更好，缓存命中率更高。
• 批量处理：将4500多万次循环缩减为几千次，大幅降低Python解释器的循环开销。
• 内置方法替代自定义函数：int.from_bytes是C实现的内置方法，比自定义的bytes2int快数倍。

内容的提问来源于stack exchange，提问作者Angelo