Python memoryview 处理大文件分片：用 readinto 降低上传缓冲区复制

批量上传归档文件的时候，大家最先优化的往往是网络传输环节，很容易忽略本地文件读取阶段反复生成临时 bytes 带来的额外开销。一个 64 MB 的文件如果按 1 MB 读取后再切片、拼接、计算校验摘要，进程的内存峰值很快就会打破「只占用单个分片缓冲区」的预期。Python 的 memoryview 可以把同一块可变缓冲区直接交给后续逻辑处理，配合 bytearray 和 readinto()，把读取、分片和校验的边界固定下来。

先看清 bytes 切片为什么会增加一次复制

假设上传器每次处理 1 MiB。直接对 bytes 做切片，得到的是一块新的字节对象；如果后面还要拼接头部、计算摘要或交给另一个缓存队列，原始块和新块会在短时间内同时存在。单次调用不一定明显，多个文件一起处理时就会表现为内存峰值明显抬高。

chunk = source.read(1024 * 1024)
payload = chunk[128:]  # bytes 切片会创建新的 bytes 对象
send_chunk(payload)

memoryview 的切片则描述同一块底层缓冲区。这里的“少复制”不是“数据永远不会复制”：网络库如果要求只读的独立 bytes，仍可能在边界处产生复制；它的价值是让应用层不必在每个处理环节主动复制。

用 readinto 把读取固定在一个可复用缓冲区

readinto() 接收一个可写的缓冲区，并返回实际写入的字节数。每轮循环只重用同一个 bytearray，上传块用 memoryview 表示。代码里的 used 是这一轮真实有效的长度，尤其要照顾文件尾部不足一个分片的情况。

from pathlib import Path

BLOCK = 1024 * 1024

def iter_blocks(path: str):
    buffer = bytearray(BLOCK)
    view = memoryview(buffer)
    with Path(path).open("rb") as source:
        while True:
            used = source.readinto(buffer)
            if used == 0:
                break
            yield view[:used]

for number, block in enumerate(iter_blocks("backup.tar")):
    send_chunk(number, block)

每次 yield 出去的对象都是同一个底层缓冲区上的视图。调用方必须在下一轮读取发生之前完成消费；如果上传函数把视图保存到异步队列，下一轮写入就可能覆盖上一块数据。

视图复用与重试队列之间要划清生命周期

同步上传可以直接消费视图；带重试的上传器则要先决定“重试数据由谁持有”。如果队列里的任务可能晚于下一次 readinto() 执行，就不要把临时视图直接放进去，而应在入队边界复制成独立的 bytes。这一次复制是有意的，因为它换来了可控的生命周期。

def make_retry_item(block: memoryview, number: int) -> tuple[int, bytes]:
    # 重试会跨过下一轮读取，必须拥有自己的数据
    return number, bytes(block)

retry_items = []
for number, block in enumerate(iter_blocks("backup.tar")):
    try:
        send_chunk(number, block)
    except TemporaryUploadError:
        retry_items.append(make_retry_item(block, number))

如果上传接口本身是同步的，直接传视图更省内存；如果接口会异步保存参数，必须查清它是否在函数返回前完成消费。这个边界比“所有地方都用 memoryview”更重要。

用小样本验证尾块和覆盖风险

不要一上来就拿几十 GB 的归档文件压测。先用 10 个字节的样本，把块大小设成 4，检查分片内容、数量以及缓冲区复用后的表现。

from pathlib import Path

sample = Path("sample.bin")
sample.write_bytes(b"0123456789")

blocks = []
for part in iter_blocks(str(sample)):
    blocks.append(bytes(part))

assert blocks == [b"0123", b"4567", b"89"]
assert b"89" not in blocks[:2]

测试里把视图转换成 bytes 再保存，是为了记录当轮结果；生产代码是否复制，仍取决于上传接口的消费时机。上线检查可以额外记录分片序号、实际长度和摘要，发现重试时序异常时更容易定位。

常见问题

memoryview 一定比 bytes 更快吗？

不一定。它主要减少应用层主动切片复制；如果后续库仍要求创建独立 bytes，复制成本仍会出现在边界处。

为什么最后一块必须使用 view[:used]？

因为 bytearray 会被循环复用，最后一轮没有覆盖的区域可能还留着上一轮内容。只取实际读取长度才能避免上传脏数据。

视图能直接放进重试队列吗？

只有在队列保证下一轮读取前完成消费时才可以。跨轮次、跨线程或异步重试，应复制成独立 bytes。

readinto 适合所有上传 SDK 吗？

它适合能接受 buffer 或 bytes-like 对象的调用链。若 SDK 会延迟读取参数，要以 SDK 文档和最小实验确认生命周期，再决定是否复制。

把复制留在真正需要持有数据的边界

固定缓冲区、readinto() 和 memoryview 解决的是读取阶段的无意复制；它们不会替上传协议补上重试、校验和幂等逻辑。先让同步消费路径复用视图，再在跨队列的持有边界明确复制，通常能同时得到较低的内存峰值和可解释的失败恢复路径。