如何在PyQt5应用中发起网络请求且不阻塞UI界面？

核心采用PyQt5原生的QThreadPool+QRunnable+自定义信号槽实现，无需引入任何第三方异步网络库，完美兼容现有运行环境，同时规避跨线程操作UI的报错问题，可直接控制并发数。

前置说明 #

原有test.ui文件无需任何修改，仅需更新test.py代码即可。

完整修改后的test.py代码 #

import sys
from datetime import datetime

import requests
from PyQt5 import QtWidgets, uic
from PyQt5.QtCore import QThreadPool, QRunnable, pyqtSignal, QObject
from requests import RequestException


class WorkerSignals(QObject):
    """自定义工作线程信号，用于跨线程传递数据"""
    # 参数说明：请求响应对象、总耗时（秒）
    finished = pyqtSignal(object, float)
    # 参数说明：异常信息
    error = pyqtSignal(str)


class RequestWorker(QRunnable):
    """网络请求工作线程，仅处理网络逻辑，不操作任何UI控件"""
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.signals = WorkerSignals()

    def run(self):
        try:
            start = datetime.now()
            headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
            r = requests.get(self.url, headers=headers)
            total_cost = (datetime.now() - start).total_seconds()
            self.signals.finished.emit(r, total_cost)
        except RequestException as e:
            self.signals.error.emit(f'Request failed with an Exception of type {type(e).__name__}')


class UI(QtWidgets.QMainWindow):
    def __init__(self):
        super(UI, self).__init__()
        uic.loadUi('test.ui', self)
        self.web = WebManager(self)
        self.issue_request_button.clicked.connect(self.issue_request)
        self.issue_click_button.clicked.connect(self.click_counter)
        self.show()

    def click_counter(self):
        self.click_count_label.setText(str(int(self.click_count_label.text()) + 1))

    def issue_request(self):
        def handle_response(r):
            print(r.status_code if r is not None else 'No successful response')
            # 如需解析JSON，直接在此处调用 r.json() 即可拿到结果做后续处理
        self.web.issue_get(self.request_url.text(), callback=handle_response)


class WebManager:
    def __init__(self, gui):
        self.gui = gui
        # 初始化线程池，设置最大并发数为2，符合需求
        self.thread_pool = QThreadPool()
        self.thread_pool.setMaxThreadCount(2)

    def issue_get(self, url_request, callback=None):
        worker = RequestWorker(url_request)
        # 绑定信号回调，所有回调函数均在主线程执行，不会出现跨线程操作UI报错
        worker.signals.finished.connect(lambda resp, cost: self._on_request_success(resp, cost, callback))
        worker.signals.error.connect(lambda err_msg: print(err_msg))
        self.thread_pool.start(worker)

    def _on_request_success(self, response, total_request_time, callback):
        # 所有UI更新操作均在主线程执行
        self.fill_request_response_info(response, total_request_time)
        if callback:
            callback(response)

    def fill_request_response_info(self, response, total_request_time):
        request = response.request
        self.gui.request_url_textbox.setText(request.method + ' ' + request.url)
        self.gui.request_headers_textbox.setText('\n'.join(k + ':' + v for k, v in request.headers.items()))
        self.gui.response_status_textbox.setText(f'{response.status_code}  {response.reason}')
        self.gui.response_elapsed_time_textbox.setText(f'{response.elapsed} / {total_request_time}s')
        self.gui.response_headers_textbox.setText('\n'.join(k + ':' + v for k, v in response.headers.items()))
        self.gui.response_body_textbox.setText(response.content.decode('utf-8'))


app = QtWidgets.QApplication(sys.argv)
window = UI()
app.exec()

实现说明 #

• 无额外依赖：所有组件均为PyQt5原生功能，兼容Python 3.7 + PyQt5环境，无需引入httpx、aiohttp等存在兼容问题的库
• 无跨线程报错：所有UI更新逻辑均在主线程执行，工作线程仅负责网络请求，通过信号槽传递结果完全符合PyQt的线程安全要求
• 并发控制简单：直接修改QThreadPool.setMaxThreadCount的参数即可调整并发数，超出并发数的请求会自动进入队列等待，无需手动实现调度逻辑
• 结果传递灵活：完整的requests响应对象会传递到回调函数中，可直接调用status_code、json()、content等属性方法满足业务需求
• 效果验证：点击「Make Request」后立即点击「Click Counter」，计数器会立即响应增加，不会被网络请求阻塞。

内容的提问来源于stack exchange，提问作者Eddie