如何用Python实现适配PyQt6的多连接并发下载管理器

核心功能实现逻辑 #

你构想的worker+monitor双线程组架构完全可行，以下是落地实现的具体方案，同时解决你提出的所有疑问：

1. 线程安全与状态控制 #

PyQt和threading不存在兼容性问题，只需遵循界面控件只能在主线程更新的规则，通过自定义信号槽传递下载数据即可。所有状态判断、数据统计都不需要读取正在下载的文件，规避了不同系统下文件读写锁的问题，我们会用线程安全的原子变量维护下载进度。

2. 分段下载实现 #

首先通过HEAD请求确认服务器支持Range分段请求，再将文件切分为32个分片，每个线程独立下载对应分片，写入文件时加锁避免冲突：

• 分片逻辑要处理总大小不能被32整除的情况，最后一个分片包含剩余所有字节
• 每个分片请求携带Range: bytes=start-end头，拿到数据后通过seek写入文件对应偏移位置

3. 暂停/恢复/停止实现 #

通过两个threading.Event对象控制下载状态：

• pause_event：默认处于触发状态，暂停时清除触发状态，所有下载线程会自动阻塞等待，恢复时重新触发即可继续下载
• stop_event：触发后所有下载线程直接退出，同时可以选择删除未完成的文件或保存分片进度用于后续断点续传

4. 数据统计与界面更新 #

监控线程每秒计算一次下载数据，通过自定义信号发送给主线程更新界面，包含你需要的所有统计项：

• 已消耗时间：当前时间减去下载开始时间
• 当前速度：最近1秒的下载字节数差值
• 最低/最高/平均速度：从速度记录列表中计算得出
• 剩余时间：未下载大小除以平均速度，转换为时分秒格式
• 进度：已下载大小除以总大小，直接赋值给PyQt的QProgressBar即可

核心代码示例 #

下载任务核心类 #

from PyQt6.QtCore import QObject, pyqtSignal, QAtomicInteger
import threading
import requests
from pathlib import Path
import time
import datetime

# 自定义信号类，用于跨线程传递界面更新数据
class DownloadSignals(QObject):
    progress = pyqtSignal(int, int, int)  # 进度百分比、已下载字节数、总字节数
    speed = pyqtSignal(float, float, float, float)  # 当前、最低、平均、最高速度，单位MB/s
    time_info = pyqtSignal(str, str, str)  # 已消耗时间、剩余时间、预计完成时间
    finished = pyqtSignal(str)  # 下载完成回调，返回保存路径
    error = pyqtSignal(str)  # 错误回调

class DownloadTask:
    STATUS_WAITING = 0
    STATUS_DOWNLOADING = 1
    STATUS_PAUSED = 2
    STATUS_STOPPED = 3
    STATUS_FINISHED = 4

    def __init__(self, url, save_path, thread_num=32):
        self.url = url
        self.save_path = Path(save_path)
        self.thread_num = thread_num
        self.signals = DownloadSignals()
        # 状态控制
        self.status = self.STATUS_WAITING
        self.pause_event = threading.Event()
        self.stop_event = threading.Event()
        self.pause_event.set()  # 默认不暂停
        # 下载统计数据
        self.total_size = 0
        self.downloaded = QAtomicInteger(0)  # 线程安全的已下载字节计数器
        self.start_time = None
        self.speed_records = []
        self.worker_threads = []
        self.monitor_thread = None
        self.file_lock = threading.Lock()  # 文件写入锁，避免多线程写冲突

    def _get_total_size(self):
        """获取文件总大小，确认是否支持分段下载"""
        try:
            resp = requests.head(self.url, timeout=10)
            resp.raise_for_status()
            if resp.headers.get("Accept-Ranges") != "bytes":
                self.signals.error.emit("服务器不支持分段下载，已降级为单线程")
                # 可在此处添加单线程下载逻辑
                return False
            self.total_size = int(resp.headers.get("content-length", 0))
            return self.total_size > 0
        except Exception as e:
            self.signals.error.emit(f"获取文件信息失败：{str(e)}")
            return False

    def _download_chunk(self, start, end):
        """下载指定字节范围的分片"""
        headers = {"Range": f"bytes={start}-{end}"}
        current_offset = start
        try:
            resp = requests.get(self.url, headers=headers, stream=True, timeout=30)
            resp.raise_for_status()
            with open(self.save_path, "r+b") as f:
                for chunk in resp.iter_content(chunk_size=1024*1024):
                    # 检查停止信号
                    if self.stop_event.is_set():
                        return
                    # 检查暂停信号，阻塞直到恢复
                    self.pause_event.wait()
                    if chunk:
                        with self.file_lock:
                            f.seek(current_offset)
                            f.write(chunk)
                        current_offset += len(chunk)
                        self.downloaded += len(chunk)
        except Exception as e:
            self.signals.error.emit(f"分片下载失败：{str(e)}")
            self.stop()

    def _monitor_loop(self):
        """监控线程，每秒统计一次下载数据并更新界面"""
        last_downloaded = 0
        while self.status not in [self.STATUS_FINISHED, self.STATUS_STOPPED]:
            if self.status == self.STATUS_PAUSED:
                time.sleep(0.5)
                continue
            # 计算时间信息
            elapsed = datetime.datetime.now() - self.start_time
            elapsed_str = str(elapsed).split(".")[0]
            # 计算速度信息
            current_downloaded = self.downloaded.load()
            current_speed = (current_downloaded - last_downloaded) / 1024 / 1024
            self.speed_records.append(current_speed)
            valid_speeds = [s for s in self.speed_records[-30:] if s > 0]  # 只保留最近30条记录
            min_speed = min(valid_speeds) if valid_speeds else 0
            max_speed = max(valid_speeds) if valid_speeds else 0
            avg_speed = sum(valid_speeds) / len(valid_speeds) if valid_speeds else 0
            # 计算剩余时间
            remaining_bytes = self.total_size - current_downloaded
            remaining_seconds = remaining_bytes / (avg_speed * 1024 * 1024) if avg_speed > 0 else 0
            remaining_str = str(datetime.timedelta(seconds=int(remaining_seconds))) if remaining_seconds > 0 else "00:00:00"
            finish_time = (self.start_time + elapsed + datetime.timedelta(seconds=int(remaining_seconds))).strftime("%Y-%m-%d %H:%M:%S")
            # 计算进度
            progress = int(current_downloaded / self.total_size * 100) if self.total_size > 0 else 0
            # 发送信号更新界面
            self.signals.progress.emit(progress, current_downloaded, self.total_size)
            self.signals.speed.emit(current_speed, min_speed, avg_speed, max_speed)
            self.signals.time_info.emit(elapsed_str, remaining_str, finish_time)
            # 判断是否下载完成
            if current_downloaded >= self.total_size:
                self.status = self.STATUS_FINISHED
                self.signals.finished.emit(str(self.save_path))
                break
            last_downloaded = current_downloaded
            time.sleep(1)

    def start(self):
        """启动下载任务"""
        if not self._get_total_size():
            return
        # 预创建对应大小的空文件
        with open(self.save_path, "wb") as f:
            f.truncate(self.total_size)
        self.status = self.STATUS_DOWNLOADING
        self.start_time = datetime.datetime.now()
        # 启动所有分片下载线程
        chunk_size = self.total_size // self.thread_num
        for i in range(self.thread_num):
            start = i * chunk_size
            end = start + chunk_size - 1 if i != self.thread_num - 1 else self.total_size - 1
            th = threading.Thread(target=self._download_chunk, args=(start, end))
            th.daemon = True
            th.start()
            self.worker_threads.append(th)
        # 启动监控线程
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()

    def pause(self):
        """暂停下载"""
        if self.status == self.STATUS_DOWNLOADING:
            self.status = self.STATUS_PAUSED
            self.pause_event.clear()

    def resume(self):
        """恢复下载"""
        if self.status == self.STATUS_PAUSED:
            self.status = self.STATUS_DOWNLOADING
            self.pause_event.set()

    def stop(self):
        """停止下载"""
        self.status = self.STATUS_STOPPED
        self.stop_event.set()
        for th in self.worker_threads:
            th.join(timeout=2)
        # 不需要保留未完成文件的话可以取消下面的注释
        # if self.save_path.exists():
        #     self.save_path.unlink()

PyQt界面集成示例 #

from PyQt6.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QProgressBar, QLabel
import sys

class DownloadWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()
        self.current_task = None

    def init_ui(self):
        layout = QVBoxLayout()
        self.progress_bar = QProgressBar()
        self.speed_label = QLabel("当前速度：0MB/s | 最低：0MB/s | 平均：0MB/s | 最高：0MB/s")
        self.time_label = QLabel("已消耗：00:00:00 | 剩余：00:00:00 | 预计完成：--")
        self.start_btn = QPushButton("开始下载")
        self.pause_btn = QPushButton("暂停下载")
        self.resume_btn = QPushButton("恢复下载")
        self.stop_btn = QPushButton("停止下载")

        layout.addWidget(self.progress_bar)
        layout.addWidget(self.speed_label)
        layout.addWidget(self.time_label)
        layout.addWidget(self.start_btn)
        layout.addWidget(self.pause_btn)
        layout.addWidget(self.resume_btn)
        layout.addWidget(self.stop_btn)
        self.setLayout(layout)

        self.start_btn.clicked.connect(self.start_download)
        self.pause_btn.clicked.connect(lambda: self.current_task.pause() if self.current_task else None)
        self.resume_btn.clicked.connect(lambda: self.current_task.resume() if self.current_task else None)
        self.stop_btn.clicked.connect(lambda: self.current_task.stop() if self.current_task else None)

    def start_download(self):
        # 替换为实际下载地址和保存路径
        url = "https://example.com/test.zip"
        save_path = "./test.zip"
        self.current_task = DownloadTask(url, save_path, thread_num=32)
        # 绑定信号更新界面
        self.current_task.signals.progress.connect(lambda p, d, t: self.progress_bar.setValue(p))
        self.current_task.signals.speed.connect(lambda c, mi, a, ma: self.speed_label.setText(f"当前速度：{c:.2f}MB/s | 最低：{mi:.2f}MB/s | 平均：{a:.2f}MB/s | 最高：{ma:.2f}MB/s"))
        self.current_task.signals.time_info.connect(lambda e, r, f: self.time_label.setText(f"已消耗：{e} | 剩余：{r} | 预计完成：{f}"))
        self.current_task.signals.finished.connect(lambda path: self.time_label.setText(f"下载完成，文件保存至：{path}"))
        self.current_task.start()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    win = DownloadWindow()
    win.resize(800, 300)
    win.show()
    sys.exit(app.exec())

注意事项 #

• 部分服务器会限制单IP并发连接数，32线程数可根据实际情况调整，避免请求被拦截
• 如果需要支持跨会话断点续传，可将已完成的分片偏移信息持久化到本地配置文件，重启软件后读取即可跳过已下载部分
• 速度记录默认只保留最近30条，避免内存占用过高，同时计算出的平均速度更贴近当前实际下载状态
• 所有文件写入操作都加了线程锁，不会出现多线程写入导致的文件损坏问题

内容的提问来源于stack exchange，提问作者Ξένη Γήινος