如何保存Zstandard stream_reader状态以实现断点续传解压缩？

核心问题分析 #

直接从任意字节位置发起新请求并初始化zstd stream_reader失败，根源在于两点：一是zstd流式解压依赖解压上下文状态，如果断点落在某个zstd帧的中间，解压器无法识别不完整的帧头部；二是stream_reader会绑定初始请求流，单纯覆盖源对象不会改变内部引用，导致无法读取新流。

可行解决方案思路 #

要实现断点续传，必须同时保存两个关键状态：

• 已下载的原始压缩字节偏移量：用于构造下次请求的Range头，从断点位置继续下载。
• zstd解压器的内部上下文状态：让解压器能无缝继续处理未完成的压缩流，包括未解析完的帧。

Python的zstd库提供了decompressobj接口，支持手动控制压缩流输入，同时允许保存和恢复解压上下文，这是实现断点续传的核心。

代码实现 #

import zstd
import requests
import json
from pathlib import Path

# 断点状态保存文件路径
STATE_FILE = Path("zstd_resume_state.json")

def save_resume_state(downloaded_bytes, decompressor_state):
    """将断点状态保存到本地文件"""
    with open(STATE_FILE, 'w', encoding='utf-8') as f:
        # 二进制状态转latin1字符串保存，避免编码丢失
        json.dump({
            "downloaded_bytes": downloaded_bytes,
            "decompressor_state": decompressor_state.decode('latin1')
        }, f)

def load_resume_state():
    """加载本地保存的断点状态"""
    if not STATE_FILE.exists():
        return 0, None
    with open(STATE_FILE, 'r', encoding='utf-8') as f:
        data = json.load(f)
        return (
            data["downloaded_bytes"],
            data["decompressor_state"].encode('latin1')
        )

def download_with_resume(url, output_path, chunk_size=1024*1024):
    """支持断点续传的zstd流式下载解压函数"""
    downloaded_bytes, saved_state = load_resume_state()
    # 以追加模式打开输出文件，保留已解压的内容
    with open(output_path, 'ab') as output_file:
        # 初始化zstd解压器，存在历史状态则恢复
        dctx = zstd.ZstdDecompressor()
        decompressor = dctx.decompressobj(saved_state) if saved_state else dctx.decompressobj()

        # 构造Range请求头，从上次中断位置继续下载
        headers = {}
        if downloaded_bytes > 0:
            headers['Range'] = f'bytes={downloaded_bytes}-'

        try:
            with requests.get(url, headers=headers, stream=True) as r:
                r.raise_for_status()
                # 遍历原始压缩流的chunk
                for raw_chunk in r.iter_content(chunk_size=chunk_size):
                    if not raw_chunk:
                        break
                    # 更新已下载的原始字节数
                    downloaded_bytes += len(raw_chunk)
                    # 解压当前chunk（可能返回空，比如chunk只是帧的一部分）
                    decompressed_chunk = decompressor.decompress(raw_chunk)
                    if decompressed_chunk:
                        output_file.write(decompressed_chunk)
                    # 每处理10个chunk自动保存一次状态，防止意外中断丢失进度
                    if downloaded_bytes % (chunk_size * 10) == 0:
                        save_resume_state(downloaded_bytes, decompressor.get_stream_state())
                # 处理解压器中剩余的未输出数据
                remaining_data = decompressor.flush()
                if remaining_data:
                    output_file.write(remaining_data)
            # 下载完成后删除断点状态文件
            if STATE_FILE.exists():
                STATE_FILE.unlink()
            print("下载解压完成")
        except KeyboardInterrupt:
            # 用户主动中断，保存当前断点
            save_resume_state(downloaded_bytes, decompressor.get_stream_state())
            print(f"已保存断点，下次将从字节位置 {downloaded_bytes} 继续")
        except Exception as e:
            print(f"执行出错：{str(e)}")
            save_resume_state(downloaded_bytes, decompressor.get_stream_state())

# 调用示例
if __name__ == "__main__":
    download_with_resume(
        url="https://example.com/your_large_zstd_file.zst",
        output_path="decompressed_result.txt"
    )

关键细节说明 #

1. 服务器兼容性：必须确保目标服务器支持Range请求（返回206 Partial Content状态码），否则断点续传无法生效。
2. 状态保存逻辑：用decompressor.get_stream_state()获取二进制解压上下文，转成latin1编码字符串后存入JSON文件，恢复时再转回二进制，避免编码丢失。
3. 规避stream_reader局限：改用decompressobj手动处理原始压缩流，既能精确跟踪已下载字节数，又能在恢复时切换新请求流而不丢失解压上下文。
4. 进度保护：定期自动保存断点状态，避免意外断电或崩溃导致进度丢失。

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