#第一种方式,借助qasync实现。使用pip install qasync安装。
from PySide6.QtWidgets import *
from PySide6.QtCore import *
from PySide6.QtGui import *
import asyncio
from qasync import QEventLoop, asyncSlot
class Form(QWidget):
def __init__(self,parent=None):
super().__init__(parent)
self.init_ui()
self.resize(400,400)
def init_ui(self):
self.btn = QPushButton("BTN",self)
self.btn.clicked.connect(self.on_click)
@asyncSlot()
async def on_click(self):
self.btn.setEnabled(False)
await asyncio.sleep(4)
self.btn.setText("hello world")
self.btn.setEnabled(True)
if __name__ == '__main__':
import sys
app = QApplication(sys.argv)
loop = QEventLoop(app)
asyncio.set_event_loop(loop)
win = Form()
win.show()
with loop:
sys.exit(loop.run_forever())
多任务实现(有bug待解决)
from PySide6.QtWidgets import *
from PySide6.QtCore import *
from PySide6.QtGui import *
import asyncio
from qasync import QEventLoop,asyncSlot
class Form(QWidget):
def __init__(self,parent=None):
super().__init__(parent)
self.init_ui()
def init_ui(self):
layout = QVBoxLayout()
self.labe1 = QLabel("函数1:0")
self.labe2 = QLabel("函数2:0")
self.labe3 = QLabel("函数3:0")
self.btn = QPushButton("启动,并发执行")
self.btn.clicked.connect(self.on_click)
layout.addWidget(self.labe1)
layout.addWidget(self.labe2)
layout.addWidget(self.labe3)
layout.addWidget(self.btn)
self.setLayout(layout)
self.resize(300,200)
@asyncSlot()
async def on_click(self):
#将main函数作为任务加入事件循环
asyncio.create_task(self.main())
async def fun1(self):
value =0
while True:
value+=1
self.labe1.setText(f"函数1:{value}")
await asyncio.sleep(1)
async def fun2(self):
value =0
while True:
value+=1
self.labe2.setText(f"函数2:{value}")
await asyncio.sleep(1)
async def fun3(self):
value =0
while True:
value+=1
self.labe3.setText(f"函数3:{value}")
await asyncio.sleep(2)
@asyncSlot()
async def main(self):
#创建任务,并传入协程对象
asyncio.create_task(self.fun1())
asyncio.create_task(self.fun2())
asyncio.create_task(self.fun3())
if __name__ == '__main__':
import sys
app = QApplication(sys.argv)
loop = QEventLoop(app)
asyncio.set_event_loop(loop)
win = Form()
win.show()
with loop:
loop.run_forever()
2 第二种方式,单独开一个线程用来做异步
import sys
import asyncio
from threading import Thread
from PySide6 import QtWidgets
from PySide6 import QtCore
class AsyncWorker(QtCore.QObject):
finished = QtCore.Signal(str) # 定义完成信号
def __init__(self):
super().__init__()
self.loop = asyncio.new_event_loop()
self.thread = Thread(target=self.run_loop, daemon=True)
self.thread.start()
def run_loop(self):
asyncio.set_event_loop(self.loop)
self.loop.run_forever()
async def _async_task(self):
await asyncio.sleep(2)
return "结果数据"
def execute_task(self):
future = asyncio.run_coroutine_threadsafe(self._async_task(), self.loop)
future.add_done_callback(lambda f: self.finished.emit(f.result()))
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
self.button = QtWidgets.QPushButton("启动任务", self)
self.button.clicked.connect(self.on_click)
self.worker = AsyncWorker()
self.worker.finished.connect(self.update_ui) # 绑定信号
self.resize(400,300)
def on_click(self):
self.button.setEnabled(False)
self.worker.execute_task()
def update_ui(self, result):
self.button.setText(result)
self.button.setEnabled(True)
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())