Python下载加速器开发：无Content-Length时如何实现分片并发下载？

问题背景 #

我正在用Python开发一款下载加速器以提升下载速度，但部分服务器的响应头中没有Content-Length字段，不清楚如何对下载内容进行分片，从而通过多个socket（计划使用select）实现并发下载。目前我已经实现了Chunked传输编码的单连接下载逻辑，相关代码如下：

1. Chunk解析函数 #

def parse_chunk(chunk: bytes, current_chunk_length_remaining: int) -> tuple[bytes, int, bytes]:
    """
        此函数接收数据块并按照RFC 2616 3.6.1规则解析
        返回值：
            real_data | current_chunk: bytes - 实际有效数据
            chunk_length_remaining: int - 剩余未接收的当前块长度
            remaining_data: bytes - 解析后剩余的未处理数据
    """
    # 首次处理块，或上一个块已在之前的数据包中处理完成
    if current_chunk_length_remaining == 0:
        chunk_header, chunk_body = chunk.split(b'\r\n', 1)
        try:
            response_length, chunk_extension = chunk_header.split(b';')
        except ValueError:
            response_length = chunk_header.split(b';')[0]
            chunk_extension = b''
        response_length = int(response_length, base=16)  # 十六进制转十进制
        real_data = chunk_body[:response_length]
        remaining_data = chunk_body[response_length:]
        if len(real_data) < response_length:  # 当前块未接收完整
            chunk_length_remaining = response_length - len(chunk_body)
        else:  # 当前块已接收完整
            chunk_length_remaining = 0
        return real_data, chunk_length_remaining, remaining_data
    # 上一个数据包未完整接收当前块
    elif current_chunk_length_remaining > 0:
        current_chunk = chunk[:current_chunk_length_remaining]
        remaining_data = chunk[current_chunk_length_remaining:]
        if len(current_chunk) < current_chunk_length_remaining:  # 当前块仍未接收完整
            chunk_length_remaining = current_chunk_length_remaining - len(current_chunk)
        else:  # 当前块已接收完整
            chunk_length_remaining = 0
        return current_chunk, chunk_length_remaining, remaining_data

2. HTTP请求解析函数 #

def parse_http_get_response(server_interface: socket, link: str, target_host: str, user_agent: str) \
        -> tuple[bytes, bytes]:
    request = "GET /" + link + \
              " HTTP/1.1\r\nHost:{}\r\nUser-Agent: {}\r\nConnection: close\r\n\r\n".format(target_host, user_agent)
    server_interface.send(request.encode('utf-8'))
    data = server_interface.recv(BUFFER_SIZE)
    http_headers, response_body = data.split(b'\r\n\r\n', 1)
    return http_headers, response_body

3. 单连接Chunk下载函数及测试代码 #

def download_chunk_data(server_interface: socket, path_to_file: str, link: str, target_host: str,
                        user_agent: str):
    _, response_body = parse_http_get_response(server_interface, link, target_host, user_agent)
    real_data, chunk_length_remaining, remaining_data = parse_chunk(response_body, 0)
    is_start = True
    with open(path_to_file, "wb") as res:
        is_file_end = False
        end_of_file = re.compile(b'\r\n0\r\n\r\n')
        data = real_data
        while True:
            if not is_start:
                real_data, chunk_length_remaining, remaining_data = parse_chunk(data, chunk_length_remaining)
                tmp = b''
                if len(remaining_data) == 0:
                    tmp = real_data
                if end_of_file.match(remaining_data):
                    tmp = real_data
                    is_file_end = True
                while chunk_length_remaining != 0 and len(remaining_data) != 0:
                    tmp += real_data
                    real_data, chunk_length_remaining, remaining_data = \
                        parse_chunk(remaining_data, chunk_length_remaining)
                data = tmp
            res.write(data)
            if not data or is_file_end:
                server_interface.close()
                break
            data = server_interface.recv(BUFFER_SIZE)
            is_start = False

targetHost = "www.livrespourtous.com"
targetPort = 80
BUFFER_SIZE = 1024
userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 " \
             "Safari/537.36 "
testLink = "http://www.livrespourtous.com/e-books/detail/Le-PDF-gratuit-et-libre/onecat/0.html"

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((targetHost, targetPort))
    download_chunk_data(client, "response.html", testLink, targetHost, userAgent)

核心问题分析 #

当服务器返回Chunked编码且无Content-Length时，无法直接使用HTTP Range请求进行字节分片下载——因为Range请求需要知道资源的总长度，而Chunked编码本身就是服务器无法预先确定响应长度的场景。因此需要换一种思路实现并发加速。

可行解决方案 #

方案1：优先尝试获取Content-Length #

先发送HEAD请求尝试获取资源的Content-Length，如果能拿到，就可以用传统的多连接Range分片下载，这是效率最高的方式：

import socket

def get_content_length(target_host, target_port, link, user_agent):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((target_host, target_port))
    request = (
        f"HEAD /{link} HTTP/1.1\r\n"
        f"Host: {target_host}\r\n"
        f"User-Agent: {user_agent}\r\n"
        f"Connection: close\r\n\r\n"
    )
    sock.send(request.encode('utf-8'))
    headers = sock.recv(4096).decode('utf-8')
    sock.close()
    for line in headers.split('\r\n'):
        if line.lower().startswith('content-length:'):
            return int(line.split(':')[1].strip())
    return None

拿到Content-Length后，就可以将文件分成N个片段，每个socket发送Range: bytes=start-end请求下载对应片段，最后合并文件。

方案2：Chunked编码下的多连接并发 fallback #

如果无法获取Content-Length，只能通过多连接并行接收整个数据流来提升速度（前提是服务器允许同一资源的多连接请求）。这里提供两种实现方式：

方式A：用select管理多连接流式解析

封装独立的Chunk解析状态，用select监听多个socket的可读事件，实时解析并写入文件：

import socket
import select
import re

class ChunkParser:
    def __init__(self):
        self.chunk_length_remaining = 0
        self.remaining_data = b''
        self.is_file_end = False
        self.end_of_file = re.compile(b'\r\n0\r\n\r\n')

    def parse(self, new_data: bytes) -> tuple[bytes, bool]:
        total_data = self.remaining_data + new_data
        self.remaining_data = b''
        output = b''

        while total_data and not self.is_file_end:
            if self.chunk_length_remaining == 0:
                if b'\r\n' not in total_data:
                    self.remaining_data = total_data
                    break
                chunk_header, chunk_body = total_data.split(b'\r\n', 1)
                try:
                    response_length_bytes, _ = chunk_header.split(b';', 1)
                except ValueError:
                    response_length_bytes = chunk_header
                response_length_bytes = response_length_bytes.strip()
                if response_length_bytes == b'0':
                    self.is_file_end = True
                    break
                response_length = int(response_length_bytes, base=16)
                if len(chunk_body) >= response_length:
                    output += chunk_body[:response_length]
                    total_data = chunk_body[response_length:]
                    if total_data.startswith(b'\r\n'):
                        total_data = total_data[2:]
                else:
                    output += chunk_body
                    self.chunk_length_remaining = response_length - len(chunk_body)
                    total_data = b''
            else:
                take = min(self.chunk_length_remaining, len(total_data))
                output += total_data[:take]
                self.chunk_length_remaining -= take
                total_data = total_data[take:]
                if self.chunk_length_remaining == 0 and total_data.startswith(b'\r\n'):
                    total_data = total_data[2:]
        self.remaining_data = total_data
        return output, self.is_file_end

def create_chunk_socket(target_host, target_port, link, user_agent):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((target_host, target_port))
    request = (
        f"GET /{link} HTTP/1.1\r\n"
        f"Host: {target_host}\r\n"
        f"User-Agent: {user_agent}\r\n"
        f"Connection: close\r\n\r\n"
    )
    sock.send(request.encode('utf-8'))
    data = sock.recv(4096)
    headers, body = data.split(b'\r\n\r\n', 1)
    if b'Transfer-Encoding: chunked' not in headers.lower():
        sock.close()
        raise ValueError("Server doesn't support chunked transfer")
    parser = ChunkParser()
    initial_data, _ = parser.parse(body)
    return sock, parser, initial_data

def concurrent_chunk_download(target_host, target_port, link, user_agent, save_path, conn_count=3):
    sockets = []
    parsers = []
    active_conns = []
    initial_datas = []

    for _ in range(conn_count):
        sock, parser, init_data = create_chunk_socket(target_host, target_port, link, user_agent)
        sockets.append(sock)
        parsers.append(parser)
        active_conns.append(sock)
        initial_datas.append(init_data)

    with open(save_path, "wb") as f:
        for data in initial_datas:
            f.write(data)

        while active_conns:
            readable, _, _ = select.select(active_conns, [], [], 5)
            if not readable:
                continue
            for sock in readable:
                idx = sockets.index(sock)
                parser = parsers[idx]
                try:
                    data = sock.recv(1024)
                    if not data:
                        active_conns.remove(sock)
                        sock.close()
                        continue
                    parsed_data, is_end = parser.parse(data)
                    if parsed_data:
                        f.write(parsed_data)
                    if is_end:
                        active_conns.remove(sock)
                        sock.close()
                except Exception as e:
                    print(f"Conn error: {e}")
                    active_conns.remove(sock)
                    sock.close()

# 测试调用
targetHost = "www.livrespourtous.com"
targetPort = 80
userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36"
testLink = "e-books/detail/Le-PDF-gratuit-et-libre/onecat/0.html"

concurrent_chunk_download(targetHost, targetPort, testLink, userAgent, "concurrent_response.html", conn_count=3)

方式B：多线程抢跑下载

启动多个线程同时下载完整资源，只要有一个线程完成就停止其他线程，避免不必要的重复下载：

import socket
import re
from threading import Thread
import queue

class ChunkParser:
    def __init__(self):
        self.chunk_length_remaining = 0
        self.remaining_data = b''
        self.is_file_end = False
        self.end_of_file = re.compile(b'\r\n0\r\n\r\n')

    def parse(self, new_data: bytes) -> tuple[bytes, bool]:
        total_data = self.remaining_data + new_data
        self.remaining_data = b''
        output = b''

        while total_data and not self.is_file_end:
            if self.chunk_length_remaining == 0:
                if b'\r\n' not in total_data:
                    self.remaining_data = total_data
                    break
                chunk_header, chunk_body = total_data.split(b'\r\n', 1)
                try:
                    response_length_bytes, _ = chunk_header.split(b';', 1)
                except ValueError:
                    response_length_bytes = chunk_header
                response_length_bytes = response_length_bytes.strip()
                if response_length_bytes == b'0':
                    self.is_file_end = True
                    break
                response_length = int(response_length_bytes, base=16)
                if len(chunk_body) >= response_length:
                    output += chunk_body[:response_length]
                    total_data = chunk_body[response_length:]
                    if total_data.startswith(b'\r\n'):
                        total_data = total_data[2:]
                else:
                    output += chunk_body
                    self.chunk_length_remaining = response_length - len(chunk_body)
                    total_data = b''
            else:
                take = min(self.chunk_length_remaining, len(total_data))
                output += total_data[:take]
                self.chunk_length_remaining -= take
                total_data = total_data[take:]
                if self.chunk_length_remaining == 0 and total_data.startswith(b'\r\n'):
                    total_data = total_data[2:]
        self.remaining_data = total_data
        return output, self.is_file_end

def download_worker(target_host, target_port, link, user_agent, result_queue, stop_flag):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((target_host, target_port))
        request = (
            f"GET /{link} HTTP/1.1\r\n"
            f"Host: {target_host}\r\n"
            f"User-Agent: {user_agent}\r\n"
            f"Connection: close\r\n\r\n"
        )
        sock.send(request.encode('utf-8'))
        data = sock.recv(4096)
        headers, body = data.split(b'\r\n\r\n', 1)
        parser = ChunkParser()
        total_data = b''
        init_data, _ = parser.parse(body)
        total_data += init_data
        while not parser.is_file_end and not stop_flag[0]:
            chunk = sock.recv(4096)
            if not chunk:
                break
            parsed_data, is_end = parser.parse(chunk)
            total_data += parsed_data
            if is_end:
                break
        if not stop_flag[0]:
            result_queue.put(total_data)
            stop_flag[0] = True
    except Exception as e:
        print(f"Worker error: {e}")
    finally:
        sock.close()

def concurrent_chunk_download_fallback(target_host, target_port, link, user_agent, save_path, conn_count=3):
    result_queue = queue.Queue()
    stop_flag = [False]
    threads = []

    for _ in range(conn_count):
        t = Thread(target=download_worker, args=(target_host, target_port, link, user_agent, result_queue, stop_flag))
        t.start()
        threads.append(t)

    final_data = result_queue.get()
    stop_flag[0] = True
    for t in threads:
        t.join()

    with open(save_path, "wb") as f:
        f.write(final_data)

# 测试调用
targetHost = "www.livrespourtous.com"
targetPort = 80
userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36"
testLink = "e-books/detail/Le-PDF-gratuit-et-libre/onecat/0.html"

concurrent_chunk_download_fallback(targetHost, targetPort, testLink, userAgent, "fallback_response.html", conn_count=3)

注意事项 #

1. 服务器兼容性：部分服务器会限制单IP的并发连接数，或对同一资源的多连接请求返回相同的数据流，此时需要调整连接数或切换到方案1。
2. 资源类型限制：Chunked编码通常用于动态生成的内容（如网页、实时数据），这类资源本身不适合分片下载，多连接加速的效果取决于服务器的带宽策略。
3. 错误处理：实际应用中需要增加更多异常处理逻辑，比如连接超时、解析失败等情况的重试机制。

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