Python threading并行下载文件内容异常问题求助

核心问题分析与修复点 #

1. 锁的作用域错误 #

原代码中每个DownloadPart实例持有独立的锁，无法实现多线程对同一文件操作的互斥。多个线程同时写入文件时，会导致文件指针混乱、内容覆盖，最终文件损坏。

修复：将锁改为全局共享，在Downloader中创建一个锁，传递给所有DownloadPart实例使用。

2. 文件未预分配空间 #

在部分操作系统（如Windows）中，直接向文件的偏移位置写入数据不会自动填充空白区域，导致文件实际大小小于预期，后续读取时内容缺失。

修复：打开文件后立即调用file.truncate(file_size)预分配与目标文件大小一致的空间。

3. tqdm进度条非线程安全 #

tqdm的update方法不是线程安全的，多线程同时调用会导致进度计数错误（虽不影响文件内容，但会导致进度显示异常）。

修复：将进度条的更新操作也放在锁的保护范围内。

4. 增加请求错误检测 #

原代码未对下载请求的响应状态做校验，若服务器返回错误状态码（如Range请求不被支持），仍会继续写入无效内容。

修复：添加response.raise_for_status()检测请求是否成功。

修复后的完整代码 #

import requests
import threading
from tqdm import tqdm

DOWNLOAD_CHUNK_SIZE = 1 << 20  # 1 MiB

class DownloadPart:
    def __init__(self, url, byte_range, global_lock) -> None:
        self.url = url
        self.byte_range = byte_range
        self.global_lock = global_lock  # 使用全局共享锁

    def download(self, file, pbar=None):
        start, end = self.byte_range
        response = requests.get(
            self.url,
            headers={"Range": f"bytes={start}-{end}"},
            allow_redirects=True,
            stream=True,
        )
        response.raise_for_status()  # 检测请求错误

        written = 0
        for chunk in response.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE):
            if chunk:
                with self.global_lock:  # 用with自动管理锁，避免遗漏释放
                    file.seek(start + written)
                    length = file.write(chunk)
                    file.flush()
                    written += length
                    pbar.update(length)


class Downloader:
    def __init__(self, url, parts=10):
        self.url = url
        self.parts = parts
        self.global_lock = threading.Lock()  # 创建全局互斥锁

    def _get_file_size(self) -> int:
        # 先尝试HEAD请求获取文件大小
        info = requests.head(self.url, allow_redirects=True)
        try:
            info.raise_for_status()
            size = info.headers.get("content-length")
            if size:
                return int(size)
        except requests.exceptions.RequestException:
            pass
        
        # HEAD失败时改用GET请求头获取
        with requests.get(self.url, stream=True, allow_redirects=True) as resp:
            resp.raise_for_status()
            size = resp.headers.get("content-length")
            assert size, "无法获取目标文件大小"
            return int(size)

    def download(self, filename):
        file_size = self._get_file_size()
        size_per_part = file_size // self.parts

        print(f"文件总大小: {file_size} 字节, 每个分片大小: {size_per_part} 字节")

        # 用with语句管理文件，自动关闭
        with open(filename, "wb") as file:
            file.truncate(file_size)  # 预分配文件空间
            pbar = tqdm(total=file_size, unit="B", unit_scale=True)

            threads = []
            for index in range(self.parts):
                start = size_per_part * index
                # 最后一个分片覆盖剩余所有字节
                end = file_size - 1 if index == self.parts - 1 else size_per_part * (index + 1) - 1
                part = DownloadPart(self.url, (start, end), self.global_lock)
                thread = threading.Thread(target=part.download, args=(file, pbar))
                thread.start()
                threads.append(thread)

            # 等待所有线程完成
            for thread in threads:
                thread.join()

            pbar.close()


# 测试URL
URL = "https://s-delivery38.mxdcontent.net/v/8a5f59673042ed97c402be84ceeb20d9.mp4?s=TfiDzO2oBLrhub_GhToCiQ&e=1676489987&_t=1676476332"

d = Downloader(URL)
d.download("video.mp4")

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