HTTP请求断点续传后zlib解压失败的原因及解决方案

核心问题定位 #

1. 致命参数错误：Range请求头设置位置完全错误
   你当前是在调用http.request拿到响应对象resp之后才给resp.headers加Range字段，Range是请求头，需要在发起GET请求时就传入request方法的headers参数才能生效。你之前的每次重连实际都是重新下载完整文件，只是本地用追加模式写文件所以最终文件哈希正确，但解压时重复喂了全量数据，直接导致解压错误。
2. 原理认知缺失：zlib解压上下文有状态依赖
   gzip底层使用的DEFLATE算法解压依赖滑动窗口内的历史数据，默认最大窗口大小为32KB（对应zlib.MAX_WBITS=15，2^15=32768字节）。你当前断线后仅回退了2KB数据，且没有重置/恢复decompressor的状态，已经处理过部分数据的decompressor再收到重传的重复数据，自然会识别为非法块。

解决方案 #

步骤1：修正Range请求头设置逻辑 #

将Range头作为参数传入http.request方法，而非修改响应头：

resp = http.request(
    'GET',
    url,
    headers={'Range': f'bytes={total_bytes_read}-'}, # 请求阶段传参才会生效
    timeout=urllib3.Timeout(connect=15, read=40),
    preload_content=False)

步骤2：新增历史滑动窗口缓冲区重建解压上下文 #

因为DEFLATE的窗口最大为32KB，只要保留最近32KB的已成功处理的压缩数据，就能在断线后重建出和断线前完全一致的解压状态：

• 新增history_buf变量，始终保留最近32KB以上的已确认压缩数据
• 每次断线后，回退32KB的下载偏移（小于0则从0开始）
• 重连后先新建decompressobj，将history_buf完整喂入新的解压对象，不处理输出，仅重建状态
• 之后再正常处理新拉取的续传数据即可

步骤3：修正状态同步逻辑 #

每次成功处理一个chunk后，更新已确认的字节偏移，同时将chunk加入history_buf，并裁剪history_buf只保留最近32KB的数据即可。

完整修改后代码 #

import urllib3
import certifi
import os
import time
import zlib

def patch_urllib3():
    """Set urllib3's enforce_content_length to True by default."""
    previous_init = urllib3.HTTPResponse.__init__
    def new_init(self, *args, **kwargs):
        previous_init(self, *args, enforce_content_length = True, **kwargs)
    urllib3.HTTPResponse.__init__ = new_init

patch_urllib3()

# 配置参数
url = "https://opendata.rapid7.com/sonar.http/2021-11-27-1638020044-http_get_8899.json.gz"
local_filename = '2021-11-27-1638020044-http_get_8899_script.json.gz'
http = urllib3.PoolManager(ca_certs=certifi.where())

# 常量定义
CHUNK_SIZE = 2048
# zlib最大滑动窗口大小32KB
WINDOW_SIZE = zlib.MAX_WBITS << 10

# 全局状态变量
total_confirmed_bytes = 0
# 历史缓冲区，仅保留最近32KB的已确认压缩数据
history_buf = b''
decompressor = zlib.decompressobj(zlib.MAX_WBITS|16)

while True:
    # 计算续传起始位置，最多回退一个窗口大小
    start_offset = max(0, total_confirmed_bytes - WINDOW_SIZE)
    print(f"发起请求，起始字节：{start_offset}")
    try:
        resp = http.request(
            'GET',
            url,
            headers={'Range': f'bytes={start_offset}-'},
            timeout=urllib3.Timeout(connect=15, read=40),
            preload_content=False
        )
        # 续传场景：重建解压上下文
        if start_offset > 0:
            decompressor = zlib.decompressobj(zlib.MAX_WBITS|16)
            # 喂入历史缓冲区重建状态，忽略输出
            decompressor.decompress(history_buf)
        # 打开文件，跳转至已确认的写入位置
        mode = 'ab' if os.path.exists(local_filename) else 'wb'
        with open(local_filename, mode) as f:
            for chunk in resp.stream(CHUNK_SIZE):
                # 跳过回退段已经处理过的字节
                if start_offset < total_confirmed_bytes:
                    skip_len = min(len(chunk), total_confirmed_bytes - start_offset)
                    chunk = chunk[skip_len:]
                    start_offset += skip_len
                    if not chunk:
                        continue
                # 写入本地文件、解压
                f.write(chunk)
                decompressed_data = decompressor.decompress(chunk)
                # 这里添加你的数据加工、写入MySQL逻辑即可
                # 更新状态
                total_confirmed_bytes += len(chunk)
                # 更新历史缓冲区，仅保留最近一个窗口大小
                history_buf += chunk
                if len(history_buf) > WINDOW_SIZE:
                    history_buf = history_buf[-WINDOW_SIZE:]
                print(f"已处理字节：{total_confirmed_bytes:,}", end='\r')
        print("\n全量数据下载解压完成")
        break
    except Exception as e:
        print(f"\n运行出错：{str(e)}")
        print(f"当前已确认字节：{total_confirmed_bytes}，30秒后重试")
        time.sleep(30)

resp.release_conn()

方案优势 #

• 仅额外占用32KB内存作为历史缓冲区，完全符合Docker资源受限场景要求
• 无需对齐gzip压缩块边界，依靠DEFLATE滑动窗口特性即可完美重建解压状态
• 修正后的Range请求逻辑真正实现断点续传，不会重复下载全量文件

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