前言

最开始写PyQt应用，一般都是单个线程（主线程）一把嗦，遇到阻塞操作难免遇到界面卡顿。在Linux和Mac下还好，Windows操作系统下就经常会提示“程序无响应”。

初级解决办法就是通过耗时操作用多线程后台处理，防止过度阻塞主线程UI响应。然而，多线程也有缺点，GIL限制甚至效率不如单线程/跨线程无法更新UI等。

中级解决办法可能会用到多进程解决问题，但是缺点也很明显，程序架构复杂度上升可维护性下降。

在Python和Go语言耳边经常听到别人鼓吹协程的优秀，查看了网上大多Python关于协程的教程后，评论区包括我一片云里雾里。

这些教程通常限制在两个领域：asyncio异步请求爬虫、aiohttp并发web服务，很不幸，这两个领域都没能解决我的实际需求。

在我的应用中，项目是基于python3和pyqt5的用户界面程序，需要无障碍支持windows/linux/mac平台。程序卡顿主要集中在大量文件拷贝（图片、视频、视频片段、音频等），大大小小一次平均超过1GB，在windows平台偶尔出现界面无响应。

最终，在一个星期有点时间就膜拜大佬们（原创or抄袭）的各种协程教程后，终于找到这样一篇文章：https://www.helplib.com/GitHub/article_112085 文中直接将程序通过asyncio启动，按钮槽函数中能够完全异步处理业务。经过测试文章中的代码支持Linux和Windows。

谨慎起见，又去quamash的pypi主页 https://pypi.org/project/Quamash/ ，找到样例代码，也是支持Linux和Windows

准备

首先，写了个简单脚本启动一个单界面，包含两个按钮：

拷贝文件（模拟阻塞操作）
弹出提示框（确认ui是否被阻塞）

import shutil
import sys
from PyQt5 import QtWidgets
def do_copy(target):
    for i in range(10):
        print(f'copy {target} at loop {i}')
        shutil.copy('C:/TMP/a.mp4', f'C:/TMP/{target}.mp4')
    return f'copy {target} done'
class Example(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()
    def init_ui(self):
        btn_copy = QtWidgets.QPushButton('Copy', self)
        btn_copy.resize(btn_copy.sizeHint())
        btn_copy.move(50, 50)
        btn_copy.clicked.connect(self.btn_copy_clicked)
        btn_alert = QtWidgets.QPushButton('Alert', self)
        btn_alert.resize(btn_alert.sizeHint())
        btn_alert.move(150, 50)
        btn_alert.clicked.connect(self.btn_alert_clicked)
        self.setGeometry(300, 300, 300, 200)
        self.show()
    def btn_copy_clicked(self, b_checked):
        for i in range(10):
            do_copy(i)
    def btn_alert_clicked(self, b_checked):
        QtWidgets.QMessageBox().information(self, '提示', 'UI响应了')
app = QtWidgets.QApplication(sys.argv)
dlg = Example()
dlg.show()
sys.exit(app.exec_())

运行后点击Copy按钮，尝试点击Alert按钮没反应，不一会就无响应了

直到拷贝完成，才响应刚才的Alert按钮点击

改造

接下来，尝试参考实例代码，修改脚本，改为异步处理按钮业务

import asyncio
import shutil
import sys
import quamash
from PyQt5 import QtWidgets
async def do_copy(target):
    for i in range(10):
        await asyncio.sleep(0)
        print(f'copy {target} at loop {i}')
        shutil.copy('C:/TMP/a.mp4', f'C:/TMP/{target}.mp4')
    return f'copy {target} done'
async def async_copy():
    for i in range(10):
        await do_copy(i)
class Example(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()
    def init_ui(self):
        btn_copy = QtWidgets.QPushButton('Copy', self)
        btn_copy.resize(btn_copy.sizeHint())
        btn_copy.move(50, 50)
        btn_copy.clicked.connect(self.btn_copy_clicked)
        btn_alert = QtWidgets.QPushButton('Alert', self)
        btn_alert.resize(btn_alert.sizeHint())
        btn_alert.move(150, 50)
        btn_alert.clicked.connect(self.btn_alert_clicked)
        self.setGeometry(300, 300, 300, 200)
        self.show()
    def btn_copy_clicked(self, b_checked):
        asyncio.ensure_future(async_copy(), loop=loop)
    def btn_alert_clicked(self, b_checked):
        QtWidgets.QMessageBox().information(self, '提示', 'UI响应了')
app = QtWidgets.QApplication(sys.argv)
loop = quamash.QEventLoop(app)
asyncio.set_event_loop(loop)  # NEW must set the event loop
with loop:
    w = Example()
    w.show()
    loop.run_forever()
print('Coroutine has ended')

运行中，就能够在一定的时间内响应UI了

分析

主要原理就是界面应该是UI的消息循环也是主线程异步的，然后拷贝过程中，通过await asyncio.sleep(0)和await do_copy(i)主动让出上下文，切换到UI消息循环，从而能够响应按钮点击。

优点

同步代码几乎无需改动
基本不阻塞UI
需要更新UI时线程安全，不会跨线程操作UI造成崩溃

缺点

大文件还是会阻塞很久
需要主动将耗时操作切分多段
async/await 传染，好在 asyncio.ensure_future 能阻断

参考链接

https://www.helplib.com/GitHub/article_112085
https://pypi.org/project/Quamash/
https://github.com/harvimt/quamash