asyncio 和 PyQt5：在 QThreadPool 中运行异步函数时结果不一致

Question

我正在开发一个 PyQt5 项目，我需要在不阻塞主 UI 的情况下调用 asyncio 函数 (

self.handler.request_data

)。 我决定在线程内运行异步函数，并且由于我将同时执行多个任务，因此我选择了 QThreadPool。

这是我整理的代码片段：

class WorkerSignals(QObject):
    results = QtCore.pyqtSignal(str, object)


class Worker(QRunnable):
    def __init__(self, ident: str, func):
        super(Worker, self).__init__()

        self.ident = ident
        self.func = func

        self.signals = WorkerSignals()

    @QtCore.pyqtSlot()
    def run(self):
        try:
            res  = asyncio.run(self.func(self.ident))
            self.signals.results.emit(self.ident, res)
        except:
            traceback.print_exc()
            self.signals.results.emit(self.ident, None)


class MainController(QObject):

    def __init__(self):
        # ...
        self.threadpool = QThreadPool()
        
    def request_data(self, ident: str):
        worker = Worker(ident, self.handler.request_data)
        worker.signals.results.connect(self.signal_testbed_response)
        self.threadpool.start(worker)

我面临的问题是，此代码仅偶尔适用于第一个请求的 ID。 异步函数（self.handler.request_data）包含一些记录器消息，但它们通常不会显示（或仅部分显示）。 看来线程内的 event_loop 可能没有完成异步函数，导致输出不一致。

对于如何解决这个问题有什么想法或建议吗？预先感谢！

（我知道有模块可以组合 Qt 事件循环和 asyncio，但我必须更改太多代码才能使用它）

Answer 1

我可以理解并理解为什么有人想在 Qt 代码中使用异步函数。信号和槽可用于许多用例，特别是当 Qt 允许我们在其他线程上使用它们时。但是，它可能会使代码在较大的项目上变得非常混乱，因为对于您启动的每个异步方法，您必须有一个连接到信号的匹配函数，以便捕获该方法生成的相应响应（即使它是一个错误） ).

异步函数基本上是具有另一个名称的协程，即可以随意暂停/恢复的函数。

知道了这个概念，我在Python中用几行代码实现了类似的方法，使用QEventLoop来等待阻塞函数终止执行。它使我们能够：

1. 在任何代码范围内异步调用阻塞方法；
2. 等待该阻塞方法执行完毕并捕获其返回值；不用说，当使用这个异步函数时，在阻塞函数完成执行之前我们不会遇到任何垃圾收集问题，因为范围没有变化；
3. 所有这些都不会冻结 GUI（因此该函数在另一个线程中执行）。

请注意，在使用此类任务时，您根本不应该在此函数内调用 GUI 方法。如果您需要更新异步函数内的 GUI（例如升级进度条，或显示更多对话框，或将文本写入控制台），您应该创建 Task 类的子类，并创建您自己的自定义信号，与您通常对 Qt + 多线程上的任何 QThread / QObject Worker 方法执行的操作相同。

这是我的实现：

from PySide2.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QDialog, QProgressBar
from PySide2.QtCore import QObject, QEventLoop, QThread, Signal
import traceback
import time

class Invoker(QObject):
    start = Signal()

class Task(QObject):
    finished = Signal()
    def __init__(self):
        super().__init__()
        self._data = None

    # Virtual function that runs on another thread
    def task(self):
        pass

    def main(self):
        try:
            self._data = self.task()
        except:
            print(traceback.format_exc())
            self._data = None
        self.finished.emit()

def AsyncTask(task):
    invoker = Invoker()
    thread = QThread()
    eloop = QEventLoop()

    thread.start()
    task.moveToThread(thread)
    task.finished.connect(eloop.quit)

    invoker.start.connect(task.main)

    invoker.start.emit()
    eloop.exec_()
    invoker.start.disconnect(task.main)

    thread.quit()
    thread.wait()

    return task._data

class LoadingDialog(QDialog):
    def __init__(self):
        super().__init__()

        pb = QProgressBar()
        pb.setMinimum(0)
        pb.setMaximum(0)
        pb.setValue(0)

        vbox = QVBoxLayout()
        vbox.addWidget(pb)
        self.setLayout(vbox)
        self.setWindowTitle('Loading...')

class Window(QWidget):
    def __init__(self):
        super().__init__()

        self.bt = QPushButton('Start Background Process')
        self.bt.clicked.connect(self.waitForTask)

        vbox = QVBoxLayout()
        vbox.addWidget(self.bt)
        self.setLayout(vbox)
        self.setWindowTitle('MRE')

    def waitForTask(self):
        func = lambda *a: [time.sleep(0.1) for i in range(10)]
        
        task = Task()
        task.task = func

        dialog = LoadingDialog()
        dialog.show()
        self.bt.setEnabled(False)

        ret = AsyncTask(task)

        self.bt.setEnabled(True)
        dialog.hide()

        print(ret)

###
app = QApplication()
win = Window()
win.show()
app.exec_()