QThreadWithReturn¶

QThreadWithReturn 是一个带返回值的 Qt 线程类，提供类似 concurrent.futures.Future 的 API，支持在 Qt 线程中执行函数并获取返回值。

类概述¶

Python

class QThreadWithReturn(QObject):
    """带返回值的 Qt 线程类"""

主要特性¶

返回值支持：可以获取线程执行函数的返回值
灵活回调机制：支持无参数、单参数、多参数的回调函数
超时控制：支持设置任务执行超时时间
优雅取消：支持优雅取消和强制终止
双模式运行：自动检测 Qt 应用环境，支持 Qt 和标准线程模式
Qt 集成：Qt 模式下自动处理事件循环，支持信号槽机制
标准模式：无 Qt 应用时使用标准 Python 线程机制
线程安全：提供线程安全的状态管理
内存安全：自动管理线程生命周期，防止内存泄漏

信号¶

信号	参数	描述
`finished_signal`	无	任务完成时发射（不论成功或失败）
`result_ready_signal`	`object`	任务成功完成时发射，携带结果

构造函数¶

Python

def __init__(
    self,
    func: Callable,
    *args,
    initializer: Optional[Callable] = None,
    initargs: tuple = (),
    thread_name: Optional[str] = None,
    **kwargs,
)

创建一个新的 QThreadWithReturn 实例。

参数¶

参数	类型	默认值	描述
`func`	`Callable`	必需	要在线程中执行的可调用对象
`*args`	`tuple`	-	传递给 `func` 的位置参数
`initializer`	`Optional[Callable]`	`None`	每个工作线程启动时调用的初始化函数
`initargs`	`tuple`	`()`	传递给 `initializer` 的参数元组
`thread_name`	`Optional[str]`	`None`	线程名称，用于调试和日志记录
`**kwargs`	`dict`	-	传递给 `func` 的关键字参数

示例¶

Python

# 基本用法
def compute_square(x):
    return x * x

thread = QThreadWithReturn(compute_square, 5)

# 带关键字参数
def greet(name, message="Hello"):
    return f"{message}, {name}!"

thread = QThreadWithReturn(greet, "Alice", message="Hi")

# 带初始化器
def init_worker():
    print("Worker thread initialized")

thread = QThreadWithReturn(
    compute_square, 10,
    initializer=init_worker,
    thread_name="SquareCalculator"
)

运行模式¶

QThreadWithReturn 支持两种运行模式，会自动检测当前环境：

Qt 模式¶

当存在 Qt 应用程序时（QApplication.instance() 不为 None）： - 使用 Qt 的 QThread 和信号槽机制 - 回调函数通过 QTimer.singleShot(0, ...) 在主线程中执行 - 支持 Qt 事件循环集成，不会阻塞 UI - 线程终止时自动调用 deleteLater()

标准模式¶

当不存在 Qt 应用程序时： - 使用标准 Python threading.Thread - 回调函数直接在当前线程中执行 - 使用 threading.Event() 进行同步 - 线程正常结束，无需特殊清理

主要方法¶

start()¶

Python

def start(self, timeout_ms: int = -1) -> None

启动线程执行任务。

参数¶

参数	类型	默认值	描述
`timeout_ms`	`int`	`-1`	超时时间（毫秒）。`<=0` 表示无超时

异常¶

RuntimeError：如果线程已在运行
TypeError：如果 timeout_ms 不是数字类型

说明¶

超时后会自动调用 cancel(force_stop=True)。

示例¶

Python

thread = QThreadWithReturn(long_running_task)

# 无超时启动
thread.start()

# 5秒超时启动
thread.start(timeout_ms=5000)

# 立即超时（1毫秒）
thread.start(timeout_ms=1)

result()¶

Python

def result(self, timeout_ms: int = -1) -> Any

获取任务执行结果。阻塞直到任务完成，如果在主线程调用，会导致界面冻结。

参数¶

参数	类型	默认值	描述
`timeout_ms`	`int`	`-1`	等待超时时间（毫秒）。`<=0` 表示无限等待

返回值¶

Any：任务的返回值

异常¶

CancelledError：如果任务被取消
TimeoutError：如果超时
Exception：任务执行时抛出的异常
TypeError：如果 timeout_ms 不是数字类型

示例¶

Python

thread = QThreadWithReturn(lambda: "Hello World")
thread.start()

try:
    # 无限等待结果
    result = thread.result()
    print(result)  # 输出: Hello World

    # 5秒超时等待
    result = thread.result(timeout_ms=5000)

except TimeoutError:
    print("任务超时")
except Exception as e:
    print(f"任务失败: {e}")

exception()¶

Python

def exception(self, timeout_ms: int = -1) -> Optional[BaseException]

获取任务执行时抛出的异常。

参数¶

参数	类型	默认值	描述
`timeout_ms`	`int`	`-1`	等待超时时间（毫秒）。`<=0` 表示无限等待

返回值¶

Optional[BaseException]：如果任务失败返回异常对象，成功返回 None

异常¶

CancelledError：如果任务被取消
TimeoutError：如果超时
TypeError：如果 timeout_ms 不是数字类型

示例¶

Python

def failing_task():
    raise ValueError("Something went wrong")

thread = QThreadWithReturn(failing_task)
thread.start()

try:
    exc = thread.exception()
    if exc:
        print(f"任务失败: {type(exc).__name__}: {exc}")
    else:
        print("任务成功完成")
except TimeoutError:
    print("等待超时")

cancel()¶

Python

def cancel(self, force_stop: bool = False) -> bool

取消线程执行。

参数¶

参数	类型	默认值	描述
`force_stop`	`bool`	`False`	如果为 `True`，强制终止线程；否则尝试优雅退出

返回值¶

bool：如果成功取消返回 True，如果线程已完成返回 False

说明¶

优雅取消需要线程内部检查 QThread.isInterruptionRequested()
强制终止可能导致资源泄漏，请谨慎使用

示例¶

Python

thread = QThreadWithReturn(long_running_task)
thread.start()

# 优雅取消
success = thread.cancel()
if success:
    print("任务已优雅取消")

# 强制终止
success = thread.cancel(force_stop=True)
if success:
    print("任务已强制终止")

wait()¶

Python

def wait(self, timeout_ms: int = -1, force_stop: bool = False) -> bool

等待任务完成。

参数¶

参数	类型	默认值	描述
`timeout_ms`	`int`	`-1`	超时时间（毫秒）。`<=0` 表示无限等待
`force_stop`	`bool`	`False`	如果为 `True`，超时后强制终止线程；否则优雅退出

返回值¶

bool：如果任务在超时前完成返回 True，否则返回 False

异常¶

TypeError：如果 timeout_ms 不是数字类型

示例¶

Python

thread = QThreadWithReturn(long_running_task)
thread.start()

# 等待5秒
if thread.wait(5000):
    print("任务完成")
else:
    print("任务仍在运行")

# 强制停止模式
if thread.wait(5000, force_stop=True):
    print("任务完成")
else:
    print("任务被强制停止")

running()¶

Python

def running(self) -> bool

检查任务是否正在运行。

返回值¶

bool：如果任务正在执行返回 True

示例¶

Python

thread = QThreadWithReturn(long_running_task)
print(thread.running())  # False

thread.start()
print(thread.running())  # True

thread.wait()
print(thread.running())  # False

done()¶

Python

def done(self) -> bool

检查任务是否已完成。

返回值¶

bool：如果任务已完成（成功、失败或取消）返回 True

示例¶

Python

thread = QThreadWithReturn(long_running_task)
print(thread.done())  # False

thread.start()
thread.wait()
print(thread.done())  # True

cancelled()¶

Python

def cancelled(self) -> bool

检查任务是否被取消。

返回值¶

bool：如果任务被取消返回 True

示例¶

Python

thread = QThreadWithReturn(long_running_task)
thread.start()

thread.cancel()
print(thread.cancelled())  # True

回调方法¶

add_done_callback()¶

Python

def add_done_callback(self, callback: Callable) -> None

添加任务成功完成后的回调函数。

参数¶

参数	类型	描述
`callback`	`Callable`	回调函数，参数数量会自动检测

说明¶

回调函数会在主线程中执行，支持以下几种形式： - 无参数：callback() - 单参数：callback(result) - 多参数：callback(a, b, c) - 当返回值是元组时自动解包

可以多次调用此方法添加多个回调，它们会按添加顺序依次执行。

示例¶

Python

def on_success():
    print("任务完成！")

def on_result(result):
    print(f"结果: {result}")

def on_multi_result(a, b, c):
    print(f"多个结果: {a}, {b}, {c}")

thread = QThreadWithReturn(lambda: "Hello")
thread.add_done_callback(on_success)  # 无参数
thread.add_done_callback(on_result)    # 单参数

# 多参数示例
thread = QThreadWithReturn(lambda: (1, 2, 3))
thread.add_done_callback(on_multi_result)  # 多参数，自动解包

# 可以添加多个回调
thread.add_done_callback(lambda x: print(f"第一个回调: {x}"))
thread.add_done_callback(lambda x: print(f"第二个回调: {x}"))

add_failure_callback()¶

Python

def add_failure_callback(self, callback: Callable) -> None

添加任务失败后的回调函数。

参数¶

参数	类型	描述
`callback`	`Callable`	回调函数

说明¶

回调函数会在主线程中执行，支持： - 无参数：callback() - 单参数：callback(exception)

失败回调只支持 0 或 1 个参数，因为异常对象只有一个。

示例¶

Python

def on_failure():
    print("任务失败！")

def on_error(error):
    print(f"错误: {error}")

thread = QThreadWithReturn(lambda: 1/0)  # 会抛出异常
thread.add_failure_callback(on_failure)  # 无参数
thread.add_failure_callback(on_error)    # 单参数

# 可以添加多个失败回调
thread.add_failure_callback(lambda: print("清理资源1"))
thread.add_failure_callback(lambda e: print(f"记录错误: {e}"))

完整使用示例¶

基础示例¶

Python

import time
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QLabel
from qthreadwithreturn import QThreadWithReturn


class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("QThreadWithReturn 示例")
        self.setGeometry(100, 100, 400, 200)

        self.button = QPushButton("开始计算", self)
        self.button.setGeometry(50, 50, 300, 40)
        self.button.clicked.connect(self.start_calculation)

        self.label = QLabel("等待计算...", self)
        self.label.setGeometry(50, 100, 300, 40)

    def start_calculation(self):
        """启动计算任务"""
        self.button.setEnabled(False)
        self.label.setText("计算中...")

        # 定义计算任务
        def fibonacci(n):
            if n <= 1:
                return n
            a, b = 0, 1
            for _ in range(2, n + 1):
                a, b = b, a + b
                time.sleep(0.1)  # 模拟计算时间
            return b

        # 创建线程
        thread = QThreadWithReturn(fibonacci, 20)

        # 添加回调
        thread.add_done_callback(self.on_success)
        thread.add_failure_callback(self.on_failure)

        # 启动线程
        thread.start()

    def on_success(self, result):
        """计算成功回调"""
        self.label.setText(f"斐波那契数列第20项: {result}")
        self.button.setEnabled(True)

    def on_failure(self, error):
        """计算失败回调"""
        self.label.setText(f"计算失败: {error}")
        self.button.setEnabled(True)


if __name__ == "__main__":
    app = QApplication([])
    window = MainWindow()
    window.show()
    app.exec()

高级示例：带超时和取消¶

Python

import time
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QLabel, QVBoxLayout
from qthreadwithreturn import QThreadWithReturn


class AdvancedExample(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("高级示例")
        self.setGeometry(100, 100, 400, 300)

        layout = QVBoxLayout()

        self.start_btn = QPushButton("开始任务")
        self.start_btn.clicked.connect(self.start_task)
        layout.addWidget(self.start_btn)

        self.cancel_btn = QPushButton("取消任务")
        self.cancel_btn.clicked.connect(self.cancel_task)
        self.cancel_btn.setEnabled(False)
        layout.addWidget(self.cancel_btn)

        self.status_label = QLabel("就绪")
        layout.addWidget(self.status_label)

        self.result_label = QLabel("")
        layout.addWidget(self.result_label)

        self.setLayout(layout)
        self._thread = None

    def start_task(self):
        """启动带超时的任务"""

        def long_task():
            for i in range(5):
                print(i)
                time.sleep(1)
            return "任务完成"

        self._thread = QThreadWithReturn(long_task)

        # 添加回调
        self._thread.add_done_callback(self.on_success)
        self._thread.add_failure_callback(self.on_failure)

        self.start_btn.setEnabled(False)
        self.cancel_btn.setEnabled(True)
        self.status_label.setText("任务执行中...")
        self.result_label.setText("")

        self._thread.start()

    def cancel_task(self):
        """取消任务"""
        if self._thread:
            success = self._thread.cancel(force_stop=True)
            if success:
                self.status_label.setText("任务已取消")
            else:
                self.status_label.setText("无法取消任务")

    def on_success(self, result):
        """任务成功"""
        self.status_label.setText("任务完成")
        self.result_label.setText(f"结果: {result}")
        self.start_btn.setEnabled(True)
        self.cancel_btn.setEnabled(False)

    def on_failure(self, error):
        """任务失败"""
        self.status_label.setText("任务失败")
        self.result_label.setText(f"错误: {error}")
        self.start_btn.setEnabled(True)
        self.cancel_btn.setEnabled(False)


if __name__ == "__main__":
    app = QApplication([])
    window = AdvancedExample()
    window.show()
    app.exec()

注意事项¶

线程安全：所有回调函数都在主线程中执行，可以安全地更新 UI
内存管理：库会自动管理线程生命周期，无需手动清理
异常处理：建议总是添加失败回调来处理可能的异常
超时设置：合理的超时设置可以防止程序无响应
强制终止：force_stop=True 可能导致资源泄漏，仅在紧急情况下使用