fry-qt6-multi-thread 0.5.0

基于 PyQt6 QThread 的多线程任务队列处理器，支持进度回报、暂停/恢复、取消与参数自动注入。

fry-qt6-multi-thread

基于 PyQt6 QThread 的多线程任务队列处理器。把耗时操作丢进后台线程执行， 让 GUI 主线程始终保持流畅，并通过 Qt 信号把进度、结果、错误安全地送回界面。

特性

• 任务队列：按先进先出顺序依次执行多个后台任务，互不阻塞 GUI。
• 进度回报：处理函数通过 progress_callback 上报进度，界面实时刷新。
• 暂停 / 恢复：基于 QWaitCondition 实现，暂停时线程真正休眠，零 CPU 消耗。
• 任务取消：可随时取消正在执行的任务。
• 参数自动注入：处理函数只要在签名里写出 task_id / task_name_str / progress_callback / pause_check_callback，即可自动获得对应值，无需手动传参。
• 线程安全：内部用 QMutex 保护共享状态。

环境要求

• Python ≥ 3.9
• PyQt6 ≥ 6.2（安装本包时会自动安装）

安装

pip install fry-qt6-multi-thread

从源码安装（开发用）：

git clone https://gitlab.com/fanrenyi33/fry_python_lib_d110_fry_qt6_multi_thread.git
cd fry_python_lib_d110_fry_qt6_multi_thread
pip install -e .

快速开始

安装完成后，运行下面这行命令即可看到完整的快速上手指南：

python -m fry_qt6_multi_thread

最小可运行示例：

from PyQt6.QtWidgets import QApplication
from fry_qt6_multi_thread import FryQt6MultiThread, FryQt6TaskStatus

app = QApplication([])
worker = FryQt6MultiThread()

# 1) 连接信号，接收任务事件
worker.result_signal.connect(lambda d: print("结果:", d["result"]))
worker.finished_signal.connect(
    lambda d: print("完成:", d["task_id"], d["status"].name))
worker.all_tasks_done_signal.connect(app.quit)

# 2) 添加任务（add_task 会自动启动后台线程）
#    handler 的普通参数通过关键字传入
worker.add_task("加法", lambda a, b: a + b, a=1, b=2)

app.exec()

处理函数与参数自动注入

任务处理函数（handler）就是一个普通的 Python 函数。它的普通参数由 add_task(name, handler, **kwargs) 的关键字参数提供；此外，只要在函数签名里 声明以下特殊参数名，本库就会自动注入对应的值：

参数名	注入内容
task_id	当前任务的整数 ID（由 add_task 返回的同一个值）
task_name_str	当前任务名称
progress_callback	进度回调：progress_callback(当前, 总数, 提示信息="")
pause_check_callback	暂停检查点：调用它会在暂停时阻塞；返回 True 表示任务应停止

示例 —— 一个支持进度回报、可暂停、可取消的下载任务：

import time

def download(url, progress_callback, pause_check_callback):
    total = 20
    for i in range(total):
        if pause_check_callback():      # 命中暂停会阻塞；被取消则返回 True
            return f"{url} 已取消"
        time.sleep(0.1)
        progress_callback(i + 1, total, f"正在下载 {url}")
    return f"{url} 下载完成"

worker.add_task("下载", download, url="http://example.com/big.zip")

信号一览

每个信号都携带一个 dict（all_tasks_done_signal 除外）：

信号	触发时机	负载字段
started_signal(dict)	任务开始执行	task_id, task_name_str
progress_signal(dict)	进度更新	task_id, index_now, total_num, task_name_str, message
result_signal(dict)	任务正常返回	task_id, task_name_str, result
finished_signal(dict)	任务结束（无论成败）	task_id, task_name_str, status
error_signal(dict)	任务抛出异常	task_id, task_name_str, error
paused_signal(bool)	暂停状态变化	True 表示已暂停
all_tasks_done_signal()	队列全部清空	无

finished_signal 负载里的 status 是 FryQt6TaskStatus 枚举： PENDING / RUNNING / COMPLETED / FAILED / CANCELLED。

方法一览

方法	说明
add_task(name, handler, **kwargs) -> int	添加任务并返回 task_id；首次调用会自动启动后台线程
pause() / resume()	暂停 / 恢复任务处理
cancel_current_task()	取消当前正在执行的任务
clear_queue()	清空尚未执行的待处理任务
stop()	停止后台线程（阻塞，直到线程真正退出）
get_queue_size() -> int	队列中待处理任务数
is_paused	属性：是否处于暂停状态

注意：在后台线程尚未启动时调用 pause() 不会生效——首次 add_task 启动线程时会清除暂停标志。暂停应在线程运行期间使用。

运行内置演示

python -m fry_qt6_multi_thread          # 显示快速上手指南
python -m fry_qt6_multi_thread demo     # 控制台演示（无需图形界面）
python -m fry_qt6_multi_thread gui      # 图形界面演示

examples/ 目录下还有两个独立示例，可直接从仓库根目录运行：

python examples/example_console.py      # 控制台用法示例
python examples/example_gui.py          # 完整 PyQt6 图形界面示例

运行测试

测试基于标准库 unittest 编写，无需额外依赖即可运行：

# 在仓库根目录执行
python -m unittest discover -s tests -v

也可以使用 pytest（需 pip install pytest）：

pytest

测试包含两类：

• 单元测试：tests/test_task_status.py、tests/test_inject_kwargs.py、 tests/test_progress_callback.py —— 校验枚举、参数注入、进度回调等纯逻辑。
• 集成测试：tests/test_integration.py —— 实际启动后台线程，覆盖任务完成、 异常、进度回报、暂停/恢复、取消、队列管理等完整生命周期。

从源码构建与发布

构建产物（sdist + wheel）：

pip install build twine
python -m build
python -m twine check dist/*

发布到 PyPI（用户名填 __token__，密码填 PyPI API Token）：

# 建议先用 TestPyPI 演练
python -m twine upload --repository testpypi dist/*

# 确认无误后再上传到正式 PyPI
python -m twine upload dist/*

Windows 用户也可以直接使用封装好的脚本：

pwsh scripts/publish.ps1 -BuildOnly   # 只构建
pwsh scripts/publish.ps1 -TestPyPI    # 构建并上传到 TestPyPI
pwsh scripts/publish.ps1              # 构建并上传到正式 PyPI

许可证

本项目基于 Apache License 2.0 发布。