第15篇：阶段三实战项目：仿照下载器实现多线程文件下载工具（完整代码）
哈喽～ 欢迎来到PyQt5系列的第15篇，也是阶段三的收官实战项目！经过前几章的学习，我们已经掌握了信号与槽、自定义信号、事件处理和多线程编程的核心知识。今天我们将这些知识点整合起来，开发一个仿照主流下载器的多线程文件下载工具，实现多任务下载、进度实时显示、暂停/继续/取消下载、下载速度计算等核心功能，彻底解决单线程下载卡顿、效率低的问题！
一、项目需求分析：对标主流下载器核心功能
我们聚焦下载器的核心实用功能，本次项目实现以下需求：
- 基础功能：支持输入下载链接、自定义保存路径、启动下载、暂停/继续下载、取消下载；
- 多任务管理：支持同时添加多个下载任务，每个任务独立运行，互不干扰；
- 进度展示：实时显示每个任务的下载进度、已下载大小、总大小、下载速度；
- 交互体验：任务列表清晰展示所有任务状态（等待中/下载中/已暂停/已完成/下载失败），操作按钮状态随任务状态动态切换；
- 线程安全：严格遵循“子线程不操作UI”原则，通过自定义信号传递数据，避免界面卡顿和程序崩溃；
- 异常处理：捕获网络异常、文件读写异常、链接无效等问题，给出明确的错误提示。
二、技术选型：整合阶段三核心知识点
本次项目严格基于阶段三所学内容，技术栈如下：
| 技术点 | 应用场景 |
|---|---|
| QThread多线程 | 每个下载任务对应一个子线程，避免阻塞主线程 |
| 自定义信号 | 子线程向主线程传递下载进度、速度、状态等数据 |
| 信号与槽绑定 | 按钮点击触发下载控制，子线程信号触发UI更新 |
| 事件处理 | 窗口关闭时安全停止所有下载线程，释放资源 |
| QTableWidget | 展示多下载任务的详细信息（链接、进度、状态等） |
| QFileDialog | 选择文件保存路径，提升用户体验 |
| 网络请求（requests库） | 分块下载文件，支持断点续传基础逻辑 |
三、项目架构设计：模块化分工
为了让代码结构清晰、易于维护，我们采用模块化设计，将项目分为3个核心部分：
- 下载线程类（DownloadThread）：继承
QThread，负责单个任务的下载逻辑，通过自定义信号传递进度和状态； - 任务管理类（TaskManager）：管理所有下载任务，实现任务的添加、删除、暂停/继续等统一控制；
- 主窗口类（DownloaderWindow）：负责界面布局、用户交互、接收子线程信号并更新UI。
核心架构逻辑：用户在主窗口添加下载任务 → 任务管理类创建对应的下载线程 → 子线程执行下载并发射信号 → 主窗口接收信号更新任务列表 → 用户通过按钮操作任务状态。
四、完整代码实现（可直接运行）
注意：需提前安装requests库（用于网络请求），执行命令：pip install requests
import sys
import time
import requests
from datetime import datetime
from PyQt5.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QTableWidget, QTableWidgetItem, QLineEdit, QPushButton,
QFileDialog, QMessageBox, QHeaderView
)
from PyQt5.QtCore import QThread, pyqtSignal, Qt, QTimer
from PyQt5.QtGui import QColor, QFont
# -------------------------- 1. 下载线程类：单个任务的下载逻辑 --------------------------
class DownloadThread(QThread):
# 自定义信号：传递下载进度、速度、状态等数据
progress_signal = pyqtSignal(int, str, str) # 进度(0-100)、已下载大小、速度
status_signal = pyqtSignal(str) # 任务状态（下载中/已暂停/已完成/失败）
finish_signal = pyqtSignal() # 任务完成信号
def __init__(self, url, save_path):
super().__init__()
self.url = url # 下载链接
self.save_path = save_path # 保存路径
self.is_paused = False # 暂停标志位
self.is_canceled = False # 取消标志位
self.chunk_size = 1024 * 1024 # 分块大小：1MB/块
self.downloaded_size = 0 # 已下载大小（字节）
self.total_size = 0 # 文件总大小（字节）
def run(self):
"""子线程核心执行方法：文件下载逻辑"""
try:
# 发送请求，获取文件大小（支持断点续传的基础）
headers = {}
if self.downloaded_size > 0:
headers["Range"] = f"bytes={self.downloaded_size}-" # 断点续传：从已下载位置开始
response = requests.get(self.url, headers=headers, stream=True, timeout=15)
self.total_size = int(response.headers.get("content-length", 0)) + self.downloaded_size
# 打开文件，追加模式写入（支持断点续传）
with open(self.save_path, "ab") as f:
self.status_signal.emit("下载中")
start_time = time.time() # 计时：用于计算下载速度
for chunk in response.iter_content(chunk_size=self.chunk_size):
# 处理暂停：暂停时阻塞循环，直到取消暂停
while self.is_paused:
time.sleep(0.1)
if self.is_canceled:
self.status_signal.emit("已取消")
self.finish_signal.emit()
return
# 处理取消：直接终止下载
if self.is_canceled:
self.status_signal.emit("已取消")
self.finish_signal.emit()
return
# 写入分块数据
f.write(chunk)
self.downloaded_size += len(chunk)
# 计算进度、已下载大小、速度
progress = int((self.downloaded_size / self.total_size) * 100) if self.total_size > 0 else 0
downloaded_str = self.format_size(self.downloaded_size)
total_str = self.format_size(self.total_size)
speed_str = self.calculate_speed(self.downloaded_size, start_time)
# 发射进度信号（每块发射一次，避免UI刷新太频繁）
self.progress_signal.emit(progress, f"{downloaded_str}/{total_str}", speed_str)
# 下载完成判断
if self.downloaded_size >= self.total_size and not self.is_canceled:
self.status_signal.emit("已完成")
elif self.is_canceled:
self.status_signal.emit("已取消")
else:
self.status_signal.emit("已暂停")
except requests.exceptions.RequestException as e:
self.status_signal.emit(f"失败：{str(e)}")
except Exception as e:
self.status_signal.emit(f"失败：{str(e)}")
finally:
self.finish_signal.emit()
def pause(self):
"""暂停下载"""
self.is_paused = not self.is_paused
status = "已暂停" if self.is_paused else "下载中"
self.status_signal.emit(status)
def cancel(self):
"""取消下载"""
self.is_canceled = True
self.is_paused = False # 取消暂停，让循环退出
def format_size(self, size):
"""字节大小格式化：转换为KB/MB/GB"""
units = ["B", "KB", "MB", "GB"]
index = 0
while size >= 1024 and index < len(units) - 1:
size /= 1024
index += 1
return f"{size:.2f} {units[index]}"
def calculate_speed(self, downloaded_size, start_time):
"""计算下载速度：单位 KB/s 或 MB/s"""
elapsed_time = time.time() - start_time
if elapsed_time <= 0:
return "0 B/s"
speed = downloaded_size / elapsed_time
return self.format_size(speed) + "/s"
# -------------------------- 2. 主窗口类：界面与交互逻辑 --------------------------
class DownloaderWindow(QMainWindow):
def __init__(self):
super().__init__()
self.init_ui()
self.download_tasks = {} # 存储下载任务：{任务索引: (线程实例, 链接, 保存路径)}
self.current_task_index = 0 # 任务索引计数器
def init_ui(self):
"""初始化界面布局"""
self.setWindowTitle("多线程文件下载工具")
self.resize(900, 600)
self.setMinimumSize(800, 500)
# 中心控件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 主布局：垂直布局
main_layout = QVBoxLayout(central_widget)
main_layout.setSpacing(15)
main_layout.setContentsMargins(20, 20, 20, 20)
# ---------- 顶部：下载任务添加区 ----------
add_task_layout = QHBoxLayout()
add_task_layout.setSpacing(10)
self.url_edit = QLineEdit()
self.url_edit.setPlaceholderText("请输入下载链接（如：https://xxx.xxx/file.zip）")
self.url_edit.setStyleSheet("padding: 8px; font-size: 14px;")
self.path_edit = QLineEdit()
self.path_edit.setPlaceholderText("请选择保存路径")
self.path_edit.setStyleSheet("padding: 8px; font-size: 14px;")
self.browse_btn = QPushButton("浏览")
self.add_btn = QPushButton("添加任务")
self.start_all_btn = QPushButton("开始所有任务")
for btn in [self.browse_btn, self.add_btn, self.start_all_btn]:
btn.setFixedSize(80, 35)
btn.setStyleSheet("font-size: 14px;")
add_task_layout.addWidget(self.url_edit)
add_task_layout.addWidget(self.path_edit)
add_task_layout.addWidget(self.browse_btn)
add_task_layout.addWidget(self.add_btn)
add_task_layout.addWidget(self.start_all_btn)
# ---------- 中部：下载任务列表（QTableWidget） ----------
self.task_table = QTableWidget()
# 列标题：索引、链接、保存路径、进度、大小、速度、状态、操作
self.task_table.setColumnCount(8)
self.task_table.setHorizontalHeaderLabels([
"任务ID", "下载链接", "保存路径", "进度", "大小", "速度", "状态", "操作"
])
# 表格样式优化
self.task_table.horizontalHeader().setStretchLastSection(True)
self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.task_table.verticalHeader().setVisible(False)
self.task_table.setEditTriggers(QTableWidget.NoEditTriggers) # 禁止编辑单元格
# ---------- 底部：状态栏 ----------
self.status_label = QLabel("就绪：可添加下载任务")
self.status_label.setStyleSheet("font-size: 12px; color: #666;")
# ---------- 添加所有控件到主布局 ----------
main_layout.addLayout(add_task_layout)
main_layout.addWidget(self.task_table)
main_layout.addWidget(self.status_label)
# ---------- 绑定信号与槽 ----------
self.browse_btn.clicked.connect(self.choose_save_path)
self.add_btn.clicked.connect(self.add_download_task)
self.start_all_btn.clicked.connect(self.start_all_tasks)
def choose_save_path(self):
"""选择文件保存路径"""
file_path, _ = QFileDialog.getSaveFileName(
self, "选择保存路径", "", "All Files (*.*)"
)
if file_path:
self.path_edit.setText(file_path)
def add_download_task(self):
"""添加下载任务到列表"""
url = self.url_edit.text().strip()
save_path = self.path_edit.text().strip()
# 输入验证
if not url:
QMessageBox.warning(self, "提示", "请输入下载链接！")
return
if not save_path:
QMessageBox.warning(self, "提示", "请选择保存路径！")
return
# 添加任务到表格
row = self.task_table.rowCount()
self.task_table.insertRow(row)
# 填充任务基本信息
self.task_table.setItem(row, 0, QTableWidgetItem(str(self.current_task_index)))
self.task_table.setItem(row, 1, QTableWidgetItem(url))
self.task_table.setItem(row, 2, QTableWidgetItem(save_path))
self.task_table.setItem(row, 3, QTableWidgetItem("0%"))
self.task_table.setItem(row, 4, QTableWidgetItem("0 B/未知"))
self.task_table.setItem(row, 5, QTableWidgetItem("0 B/s"))
self.task_table.setItem(row, 6, QTableWidgetItem("等待中"))
# 添加操作按钮（暂停/继续/取消）
btn_layout = QHBoxLayout()
start_btn = QPushButton("开始")
pause_btn = QPushButton("暂停")
cancel_btn = QPushButton("取消")
for btn in [start_btn, pause_btn, cancel_btn]:
btn.setFixedSize(60, 30)
btn_layout.addWidget(btn)
btn_widget = QWidget()
btn_widget.setLayout(btn_layout)
self.task_table.setCellWidget(row, 7, btn_widget)
# 绑定按钮信号
start_btn.clicked.connect(lambda: self.start_single_task(row))
pause_btn.clicked.connect(lambda: self.pause_single_task(row))
cancel_btn.clicked.connect(lambda: self.cancel_single_task(row))
# 存储任务信息
self.download_tasks[row] = {
"thread": None,
"url": url,
"save_path": save_path,
"start_btn": start_btn,
"pause_btn": pause_btn,
"cancel_btn": cancel_btn
}
# 更新状态
self.status_label.setText(f"已添加任务 {self.current_task_index}：{url}")
self.current_task_index += 1
# 清空输入框
self.url_edit.clear()
self.path_edit.clear()
def start_single_task(self, row):
"""启动单个下载任务"""
task = self.download_tasks.get(row)
if not task:
return
# 创建下载线程
if task["thread"] is None:
task["thread"] = DownloadThread(task["url"], task["save_path"])
# 绑定线程信号
task["thread"].progress_signal.connect(lambda p, s, sp: self.update_task_progress(row, p, s, sp))
task["thread"].status_signal.connect(lambda st: self.update_task_status(row, st))
task["thread"].finish_signal.connect(lambda: self.on_task_finished(row))
# 启动线程
task["thread"].start()
task["start_btn"].setEnabled(False)
task["pause_btn"].setEnabled(True)
task["cancel_btn"].setEnabled(True)
self.status_label.setText(f"任务 {row} 开始下载...")
def pause_single_task(self, row):
"""暂停/继续单个任务"""
task = self.download_tasks.get(row)
if task and task["thread"] and task["thread"].isRunning():
task["thread"].pause()
def cancel_single_task(self, row):
"""取消单个任务"""
task = self.download_tasks.get(row)
if task and task["thread"]:
task["thread"].cancel()
task["start_btn"].setEnabled(False)
task["pause_btn"].setEnabled(False)
task["cancel_btn"].setEnabled(False)
def start_all_tasks(self):
"""启动所有等待中的任务"""
for row in range(self.task_table.rowCount()):
status = self.task_table.item(row, 6).text()
if status == "等待中":
self.start_single_task(row)
def update_task_progress(self, row, progress, size_str, speed_str):
"""更新单个任务的进度、大小、速度"""
self.task_table.setItem(row, 3, QTableWidgetItem(f"{progress}%"))
self.task_table.setItem(row, 4, QTableWidgetItem(size_str))
self.task_table.setItem(row, 5, QTableWidgetItem(speed_str))
# 进度条样式（可选：设置单元格背景色）
progress_item = self.task_table.item(row, 3)
if progress < 50:
progress_item.setBackground(QColor(255, 240, 240))
else:
progress_item.setBackground(QColor(240, 255, 240))
def update_task_status(self, row, status):
"""更新单个任务的状态"""
self.task_table.setItem(row, 6, QTableWidgetItem(status))
status_item = self.task_table.item(row, 6)
if status == "下载中":
status_item.setForeground(QColor(34, 139, 34))
elif status == "已暂停":
status_item.setForeground(QColor(255, 165, 0))
elif status == "已完成":
status_item.setForeground(QColor(0, 128, 0))
elif status == "已取消":
status_item.setForeground(QColor(128, 128, 128))
else:
status_item.setForeground(QColor(220, 20, 60))
def on_task_finished(self, row):
"""任务完成后的清理工作"""
task = self.download_tasks.get(row)
if task:
task["start_btn"].setEnabled(False)
task["pause_btn"].setEnabled(False)
self.status_label.setText(f"任务 {row} 已完成！")
def closeEvent(self, event):
"""窗口关闭时安全停止所有线程"""
reply = QMessageBox.question(
self, "关闭确认", "确定要关闭下载工具吗？正在下载的任务将被取消！",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No
)
if reply == QMessageBox.Yes:
# 停止所有正在运行的线程
for row, task in self.download_tasks.items():
if task["thread"] and task["thread"].isRunning():
task["thread"].cancel()
task["thread"].wait()
event.accept()
else:
event.ignore()
if __name__ == "__main__":
app = QApplication(sys.argv)
window = DownloaderWindow()
window.show()
sys.exit(app.exec_())五、核心功能解析
1. 多线程下载核心逻辑
- 分块下载：通过
response.iter_content(chunk_size=1MB)分块读取文件，避免一次性加载大文件到内存； - 断点续传基础：利用HTTP的
Range请求头，记录已下载大小，暂停后重新下载时从断点开始； - 线程安全控制：通过
is_paused和is_canceled两个标志位控制下载循环，避免强制终止线程导致资源泄漏。
2. 自定义信号的应用
子线程定义了3个核心信号，实现与主线程的安全通信：
progress_signal：传递下载进度、已下载大小、下载速度，用于实时更新表格；status_signal：传递任务状态，用于更新状态列的文字和颜色；finish_signal：任务完成信号，用于禁用操作按钮，清理资源。
3. 任务列表的动态管理
- 表格布局：用
QTableWidget展示所有任务的详细信息，支持多列拉伸，适配不同窗口大小； - 操作按钮：每个任务行都有独立的“开始/暂停/取消”按钮，通过
lambda表达式传递任务索引，实现精准控制； - 样式优化：根据任务状态设置不同的文字颜色（如完成绿色、失败红色），进度单元格根据进度设置背景色，提升视觉体验。
4. 异常处理机制
- 网络异常：捕获
requests.exceptions.RequestException，包括超时、链接无效、服务器错误等； - 文件异常：捕获文件读写时的权限错误、路径不存在等问题；
- 用户操作异常：避免重复点击开始按钮、暂停非运行中的任务等无效操作。
六、运行与测试步骤
- 环境准备：确保已安装
PyQt5和requests库； - 运行代码：将代码保存为
multi_thread_downloader.py，执行命令：python multi_thread_downloader.py； 添加任务：
- 输入一个有效的下载链接（如Python官网的安装包：
https://www.python.org/ftp/python/3.12.0/python-3.12.0-amd64.exe）； - 点击“浏览”选择保存路径（如
D:/python.exe）； - 点击“添加任务”，任务将出现在列表中，状态为“等待中”；
- 输入一个有效的下载链接（如Python官网的安装包：
测试功能：
- 点击“开始”按钮启动下载，观察进度、大小、速度是否实时更新；
- 点击“暂停”按钮，任务状态变为“已暂停”，再次点击恢复下载；
- 点击“取消”按钮，任务状态变为“已取消”，停止下载；
- 测试多任务下载：添加多个任务，点击“开始所有任务”，观察多个任务是否并行运行。
七、常见问题排查
1. 下载失败：提示“ConnectionError”
- 原因：下载链接无效、网络断开或服务器拒绝访问；
- 解决方案：检查链接是否正确，确保网络通畅，尝试更换下载链接（如使用公开的文件下载链接）。
2. 进度不更新：任务状态一直显示“下载中”
- 原因：文件过小，下载速度过快，进度直接跳到100%；或服务器不支持分块下载；
- 解决方案：尝试下载较大的文件（如几百MB的安装包），观察进度变化；检查
response.headers是否包含content-length字段。
3. 暂停后继续下载，文件损坏
- 原因：未使用追加模式写入文件，暂停后重新下载覆盖了原有文件；
- 解决方案：确保文件打开模式为
ab（追加二进制模式），而非wb（写入二进制模式）。
4. 窗口关闭后，下载线程仍在运行
- 原因：窗口关闭时未停止所有线程，子线程后台继续运行；
- 解决方案：在
closeEvent中遍历所有任务，调用cancel()方法停止线程，并调用wait()等待线程结束。
八、功能拓展思路（进阶方向）
基于当前代码，可拓展以下实用功能，进一步提升工具的实用性：
- 断点续传优化：将已下载大小保存到本地配置文件，工具重启后可继续未完成的任务；
- 多线程分块下载：将一个文件分成多个块，每个块用一个子线程下载，提升下载速度；
- 任务排序与筛选：支持按状态（如“已完成”“下载中”）筛选任务，按下载速度排序；
- 下载限速：通过控制分块下载的间隔时间，实现下载速度限制；
- 文件校验：下载完成后，计算文件的MD5值，与官方提供的MD5对比，验证文件完整性。
总结
本次实战项目完美整合了阶段三的核心知识点——多线程编程、自定义信号、信号与槽绑定、事件处理，实现了一个功能完整的多线程下载工具。通过这个项目，你不仅掌握了PyQt5多线程开发的核心逻辑，还理解了“主线程管UI，子线程做任务”的分工原则，以及如何通过信号与槽实现线程间的安全通信。
下一章我们将进入阶段四的学习——PyQt5样式表（QSS），通过类似CSS的语法，给我们的界面进行美化，打造出高颜值的现代化GUI程序！如果在项目实操中遇到问题，或者有拓展功能的想法，欢迎在评论区留言讨论～
