1. 基础知识

1.1 简介

1.1.1 多线程与多进程的概念

• 多线程 (Multithreading)

  • 定义：多线程是在单个进程内运行多个线程，每个线程可以执行不同的任务。线程是操作系统调度的基本单位。
  • 共享内存空间：线程共享同一个进程的内存空间，因此可以轻松地共享数据，但这也带来了线程安全的问题。
  • 轻量级：线程比进程更轻量级，创建和销毁的开销较小。
  • 适用场景：适用于I/O密集型任务，如文件读写、网络请求等，因为这些任务往往在等待I/O操作完成时会阻塞线程，但其他线程可以继续执行。

• 多进程 (Multiprocessing)

  • 定义：多进程是在操作系统内同时运行多个进程，每个进程拥有独立的内存空间和资源。
  • 独立内存空间：进程之间不共享内存空间，这使得它们之间的数据共享需要通过进程间通信（IPC）机制，如管道、消息队列等。
  • 重量级：进程比线程更重量级，创建和销毁的开销较大。
  • 适用场景：适用于CPU密集型任务，如复杂计算、数据处理等，因为每个进程可以独立运行在多核CPU上，充分利用多核处理能力。

1.1.2 多线程与多进程的区别

• 资源使用：
  • 多线程：共享同一进程的内存空间和资源，线程间通信（如通过全局变量或共享对象）更加便捷，但需要处理线程同步问题。
  • 多进程：每个进程有独立的内存空间和资源，进程间通信相对复杂，但能提供更好的隔离性和稳定性。
• 开销：
  • 多线程：创建和销毁的开销较小，适合轻量级的并发任务。
  • 多进程：创建和销毁的开销较大，但适合需要隔离资源和独立运行的重任务。
• 安全性：
  • 多线程：由于共享内存空间，线程安全（如竞争条件、死锁）是一个重要问题，需要使用锁、信号量等机制。
  • 多进程：各进程独立运行，安全性较高，不会出现线程间资源竞争的问题。
• 性能：
  • 多线程：适合I/O密集型任务，可以提高程序的响应速度和处理效率。
  • 多进程：适合CPU密集型任务，可以充分利用多核CPU的计算能力，提高计算效率。

1.1.3 应用场景

• 多线程应用场景

  • GUI应用：在GUI应用中，多线程可以用来处理后台任务（如文件下载、数据加载），以避免阻塞主线程，使界面保持响应。

  • 网络应用：网络服务器、爬虫等需要处理大量I/O操作的应用，可以使用多线程来处理多个客户端连接或请求。

• 多进程应用场景

  • 数据处理：在数据分析、科学计算等需要大量CPU计算的任务中，多进程可以显著提高计算速度。

  • 独立任务：需要运行彼此独立的任务（如不同的子进程执行不同的任务），避免相互干扰，提高程序的健壮性和稳定性。

1.2 Python标准库中的线程与进程库

Python标准库中的threading、multiprocessing模块的使用方法。

1.2.1 threading 模块

• 基本概念

  • 线程：threading.Thread类用于创建和管理线程。

  • 锁：threading.Lock类用于线程同步，防止竞争条件。

  • 事件：threading.Event类用于线程间通信和同步。

  • 条件变量：threading.Condition类用于更复杂的线程同步。

  • 信号量：threading.Semaphore类用于控制线程并发数量。

• 使用方法

  • 创建线程：

    • 使用threading.Thread类创建并启动线程。

    import threading  # 导入 threading 模块以使用线程功能def worker():# 定义一个线程将要执行的函数print("Thread is working")# 创建一个线程对象，指定线程执行的目标函数为 worker
    thread = threading.Thread(target=worker)# 启动线程，开始执行 worker 函数
    thread.start()# 等待线程结束，在此期间主线程将被阻塞，直到这个线程完成
    thread.join()
    

    • 调用 start() 方法后，线程将运行并执行 worker 函数。启动后，线程会在后台独立运行，不会阻塞主线程。
    • join() 方法会阻塞主线程，直到调用它的线程执行完毕。在 thread.join() 之前，主线程和新线程是并行执行的。join() 确保主线程在新线程完成之前不会继续执行。

    • 当你运行这段代码时，输出将是：

      Thread is working
      

  • 线程同步：

    • 使用threading.Lock进行线程同步，防止多个线程同时访问共享资源。

    import threading  # 导入 threading 模块以使用线程功能# 创建一个锁对象，用于确保对共享资源的安全访问
    lock = threading.Lock()# 定义一个共享资源，所有线程都会访问和修改它
    shared_resource = 0def increment():global shared_resource  # 声明使用全局变量 shared_resourcewith lock:  # 使用 with 语句来自动获取和释放锁shared_resource += 1  # 安全地递增共享资源# 创建一个包含 100 个线程对象的列表，每个线程的目标函数都是 increment
    threads = [threading.Thread(target=increment) for _ in range(100)]# 启动所有线程
    for t in threads:t.start()# 等待所有线程完成
    for t in threads:t.join()# 打印共享资源的值，期望输出 100
    print(shared_resource)  # 期望输出100

    • lock = threading.Lock() 这里Lock 对象 lock，用于确保对共享资源的安全访问。
    • 锁是一种同步原语，用于确保在同一时刻只有一个线程可以访问共享资源。
    • 定义一个共享资源 shared_resource，初始值为 0。
    • 所有线程都会访问和修改这个变量。
    • increment 函数是线程的目标函数，用于递增共享资源。
    • 使用 global 关键字声明 shared_resource 为全局变量。
    • 使用 with lock: 获取锁，并在代码块执行完毕后自动释放锁。这样可以确保只有一个线程在同一时刻修改 shared_resource。
    • 第一个for循环，遍历线程列表，启动每个线程。每个线程开始执行 increment 函数，每个线程安全地递增了 shared_resource
    • 第二个for循环遍历线程列表，调用每个线程的 join() 方法，等待线程完成。
    • join() 方法会阻塞主线程，直到调用它的线程执行完毕。
    • 上面例子的输出结果是100，这表示 100 个线程安全地递增了 shared_resource，每个线程递增一次，总计递增 100 次，确保输出结果是 100。锁的使用确保了线程在递增共享资源时不会发生竞争条件。

  • 线程间通信：

    • 使用threading.Event实现线程间的简单通信。

    import threading  # 导入 threading 模块以使用线程功能# 创建一个 Event 对象，用于线程间通信
    event = threading.Event()def waiter():# 定义一个线程将要执行的函数print("Waiting for event\n")  # 打印消息，表示线程正在等待事件event.wait()  # 等待事件被设置（即 set() 被调用），阻塞线程print("Event received")  # 事件被设置后，打印消息，表示事件已接收# 创建一个线程对象，指定线程执行的目标函数为 waiter
    thread = threading.Thread(target=waiter)# 启动线程，开始执行 waiter 函数
    thread.start()# 主线程继续执行
    print("Main thread setting event")  # 打印消息，表示主线程即将设置事件
    event.set()  # 设置事件，解除所有等待该事件的线程的阻塞状态

    • 在上面例子中：

    • 创建一个 Event 对象 event，用于在线程间通信和同步。

    • Event 对象可以在线程之间发送信号。一个线程可以等待某个事件的发生，而另一个线程可以触发这个事件。

    • waiter 函数是线程的目标函数。

    • 首先打印 "Waiting for event" 表示线程进入等待状态。

    • 调用 event.wait() 方法，该方法会阻塞线程，直到事件被设置。

    • 当事件被设置时，打印 "Event received" 表示事件已接收，线程继续执行。

    • 创建一个新的 Thread 对象，并将其目标函数设置为 waiter。

    • target=waiter 表示线程启动时会调用 waiter 函数。

    • 启动线程。调用 start() 方法后，线程将运行并执行 waiter 函数。

    • 线程开始执行，打印 "Waiting for event" 并进入阻塞状态，等待事件被设置。

    • 主线程继续执行并打印 "Main thread setting event"，表示即将设置事件。

    • 调用 event.set() 方法，设置事件。这将解除所有等待该事件的线程的阻塞状态，使它们可以继续执行。

    • 当你运行这段代码时，输出将是：

    • Waiting for event
      Main thread setting event
      Event received
      

      1. 线程启动后：waiter 函数打印 "Waiting for event" 并调用 event.wait() 进入阻塞状态。
      2. 主线程继续执行：打印 "Main thread setting event" 并调用 event.set() 设置事件。
      3. 事件被设置后：阻塞的线程解除阻塞，打印 "Event received" 并继续执行。

      通过这种方式，可以实现线程间的同步，使一个或多个线程等待某个事件的发生，而另一个线程可以触发这个事件。

1.2.2 总结

• 线程：使用threading.Thread创建和管理线程，适用于I/O密集型任务。使用Lock、Event等工具进行线程同步和通信。

2. PyQt5的多线程

2.1 QThread类

2.1.1. QThread 类的工作原理

• QThread对象：QThread类本身代表一个线程对象，可以启动、执行和管理线程的生命周期。
• 事件循环：每个QThread对象都有一个事件循环，这个循环在调用start()方法后运行。事件循环允许线程处理事件和信号。
• 线程与对象：在PyQt中，可以将对象移到某个QThread中，这样对象的槽函数将在这个线程的上下文中执行，而不是主线程中。

2.1.2 QThread 类的使用方法

• 基本用法

  • 创建自定义线程类：

    通过继承QThread类并重载run()方法来定义一个新的线程。

  from PyQt5.QtCore import QThread, pyqtSignal  # 导入必要的模块class WorkerThread(QThread):# 定义一个 pyqtSignal 对象 progress，用于发射整数类型信号progress = pyqtSignal(int)def __init__(self):super().__init__()  # 调用父类的构造函数def run(self):for i in range(100):self.sleep(1)  # 模拟长时间任务，让线程休眠1秒钟self.progress.emit(i)  # 发射进度信号，参数为当前进度值

  • 定义了一个名为 progress 的信号对象，类型为整数。
  • 这个信号将用于在线程执行过程中发射进度信息。
  • run 方法是 QThread 类的一个虚拟函数，在调用线程的 start 方法后自动被调用。
  • 在这个方法中，通过一个循环模拟一个长时间的任务。
  • 每次循环迭代时，线程会休眠1秒钟，然后发射一个进度信号，传递当前进度值 i。
  • 当创建 WorkerThread 实例并调用 start 方法时，run 方法将在一个单独的线程中执行。在这个方法中，通过循环模拟一个耗时的任务，每次循环迭代都会休眠1秒钟，然后发射一个进度信号，通知主线程当前的进度。这样，主线程就能够实时获取到后台线程的执行情况，从而更新用户界面或执行其他操作。

  • 在主线程中使用自定义线程类：

    创建线程实例并连接信号和槽函数。

  from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QLabel
  from PyQt5.QtCore import Qt, QThread, pyqtSignalclass WorkerThread(QThread):# 定义一个 pyqtSignal 对象 progress，用于发射整数类型信号progress = pyqtSignal(int)def __init__(self):super().__init__()  # 调用父类的构造函数def run(self):for i in range(100):self.sleep(1)  # 模拟长时间任务，让线程休眠1秒钟self.progress.emit(i)  # 发射进度信号，参数为当前进度值class MainWindow(QMainWindow):def __init__(self):super().__init__()self.initUI()def initUI(self):# 创建标签和按钮self.label = QLabel('Progress: 0', self)self.label.setAlignment(Qt.AlignCenter)  # 设置标签文本居中对齐self.button = QPushButton('Start', self)self.button.clicked.connect(self.start_thread)  # 将按钮点击事件连接到 start_thread 方法# 创建垂直布局，并将标签和按钮添加到布局中layout = QVBoxLayout()layout.addWidget(self.label)layout.addWidget(self.button)# 创建容器窗口，将布局设置为容器的布局container = QWidget()container.setLayout(layout)# 将容器设置为主窗口的中心窗口self.setCentralWidget(container)self.resize(500, 300)# 创建 WorkerThread 实例，并连接进度信号到更新标签的方法self.thread = WorkerThread()self.thread.progress.connect(self.update_label)# 启动线程的方法def start_thread(self):self.thread.start()# 更新标签文本的方法，接收进度值并更新标签文本def update_label(self, value):self.label.setText(f'Progress: {value}')if __name__ == '__main__':# 创建应用程序实例app = QApplication([])# 创建主窗口实例window = MainWindow()# 显示主窗口window.show()# 运行应用程序事件循环app.exec_()
  

  运行机制

  当运行这段代码时，会创建一个 PyQt5 应用程序，并显示一个窗口。窗口中包含一个标签和一个按钮。当点击按钮时，会启动一个线程，在后台执行一个任务，并实时更新标签中的进度值。这样，用户可以通过界面的交互操作来控制和监控后台任务的执行情况。运行结果如下：

  

• 高级用法

  • 使用moveToThread将对象移到另一个线程：

    将一个对象移到一个新的线程中，以便它的槽函数在这个线程中执行。

  import sys
  from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QLabel
  from PyQt5.QtCore import Qt, QThread, pyqtSignal, QObject# 定义一个工作线程类，继承自QObject
  class Worker(QObject):# 定义两个信号，用于在工作线程中发射信号finished = pyqtSignal()progress = pyqtSignal(int)# 定义一个长时间任务的方法def long_task(self):# 循环100次，模拟一个长时间的任务for i in range(100):QThread.sleep(1)  # 每次循环休眠1秒，模拟长时间任务self.progress.emit(i)  # 发射进度信号，传递当前进度值self.finished.emit()  # 任务完成后，发射完成信号# 定义一个主窗口类，继承自QMainWindow
  class MainWindow(QMainWindow):def __init__(self):super().__init__()self.initUI()def initUI(self):# 创建标签和按钮self.label = QLabel('Progress: 0', self)self.label.setAlignment(Qt.AlignCenter)self.button = QPushButton('Start', self)self.button.clicked.connect(self.start_worker)  # 将按钮点击事件连接到启动工作线程的方法# 创建垂直布局，并将标签和按钮添加到布局中layout = QVBoxLayout()layout.addWidget(self.label)layout.addWidget(self.button)# 创建容器窗口，将布局设置为容器的布局container = QWidget()container.setLayout(layout)self.setCentralWidget(container)  # 将容器设置为主窗口的中心窗口self.resize(500, 300)  # 设置窗口大小为500x300# 创建工作线程对象和线程对象self.worker = Worker()self.thread = QThread()self.worker.moveToThread(self.thread)  # 将工作线程移动到线程中# 连接工作线程的信号与槽函数self.worker.progress.connect(self.update_label)  # 将工作线程的进度信号连接到更新标签的方法self.worker.finished.connect(self.thread.quit)  # 将工作线程的完成信号连接到线程的退出方法self.thread.started.connect(self.worker.long_task)  # 将线程的启动信号连接到工作线程的长时间任务方法self.thread.finished.connect(self.thread.deleteLater)  # 将线程的完成信号连接到线程的deleteLater方法# 启动工作线程的方法def start_worker(self):self.thread.start()# 更新标签文本的方法，接收进度值并更新标签文本def update_label(self, value):self.label.setText(f'Progress: {value}')if __name__ == '__main__':app = QApplication(sys.argv)  # 创建应用程序实例window = MainWindow()  # 创建主窗口实例window.show()  # 显示主窗口sys.exit(app.exec_())  # 运行应用程序事件循环

• 总结

  • 创建自定义线程：通过继承QThread并重载run()方法可以创建自定义线程。

  • 信号与槽：在QThread中，可以使用信号和槽来与主线程进行通信，确保线程安全地更新UI。

  • 对象移到线程：通过moveToThread()方法，可以将对象移动到另一个线程，以便其槽函数在该线程中执行。

2.2 QRunnable与QThreadPool

在PyQt5中，QRunnable和QThreadPool提供了一种更高效的方式来管理和执行多线程任务，特别是当需要同时执行多个独立任务时。

2.2.1 QRunnable 与 QThreadPool 概念

• QRunnable：QRunnable是一个可运行的任务对象，需要重载其run()方法来定义具体的任务逻辑。

• QThreadPool：QThreadPool是一个线程池管理器，用于管理和执行QRunnable任务。它可以重用线程，从而减少创建和销毁线程的开销。

2.2.2 使用方法

• 定义任务 (QRunnable)

  首先，需要创建一个继承自QRunnable的类，并重载其run()方法来定义任务逻辑。

from PyQt5.QtCore import QRunnable, pyqtSlot# 定义一个任务类，继承自 QRunnable
class Task(QRunnable):def __init__(self, n):# 调用父类的构造函数super().__init__()# 初始化任务的编号self.n = n@pyqtSlot()# run 方法将在任务运行时被调用def run(self):# 打印任务开始的信息print(f"Task {self.n} is running")# 模拟一个长时间任务import timetime.sleep(2)  # 让当前线程休眠2秒# 打印任务完成的信息print(f"Task {self.n} is complete")

• 在上面的代码中，
• 导入 QRunnable 类，用于定义可运行的任务。
• 导入 pyqtSlot 装饰器，用于标记槽函数。
• 创建一个继承自 QRunnable 的子类 Task，用于表示一个可运行的任务。
• 构造函数接收一个参数 n，表示任务的编号。
• 调用父类 QRunnable 的构造函数进行初始化。
• 将任务编号 n 保存在实例变量 self.n 中。
• 使用 @pyqtSlot() 装饰器标记 run 方法为槽函数，这样可以确保它在适当的线程中执行。
• run 方法是任务执行的入口点，当任务被运行时调用。
• 打印一条消息，指示任务开始执行。
• 导入 time 模块并调用 time.sleep(2)，模拟一个长时间任务，让当前线程休眠2秒钟。
• 打印一条消息，指示任务完成。

• 使用 QThreadPool 管理和执行任务

  在主线程中创建并管理线程池，将任务添加到线程池中以执行。

from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget
from PyQt5.QtCore import QThreadPool
from PyQt5.QtCore import QRunnable, pyqtSlot# 定义一个任务类，继承自 QRunnable
class Task(QRunnable):def __init__(self, n):# 调用父类的构造函数super().__init__()# 初始化任务的编号self.n = n@pyqtSlot()# run 方法将在任务运行时被调用def run(self):# 打印任务开始的信息print(f"Task {self.n} is running")# 模拟一个长时间任务import timetime.sleep(2)  # 让当前线程休眠2秒# 打印任务完成的信息print(f"Task {self.n} is complete")# 定义主窗口类，继承自QMainWindow
class MainWindow(QMainWindow):def __init__(self):super().__init__()self.initUI()# 初始化用户界面def initUI(self):# 创建一个按钮，文本为"Start Tasks"self.button = QPushButton('Start Tasks', self)# 将按钮的点击信号连接到 start_tasks 方法self.button.clicked.connect(self.start_tasks)# 创建垂直布局，并将按钮添加到布局中layout = QVBoxLayout()layout.addWidget(self.button)# 创建一个容器窗口，将布局设置为容器的布局container = QWidget()container.setLayout(layout)# 将容器设置为主窗口的中心窗口self.setCentralWidget(container)# 创建一个线程池self.thread_pool = QThreadPool()# 启动任务的方法def start_tasks(self):# 创建并启动5个任务for i in range(5):task = Task(i)self.thread_pool.start(task)if __name__ == '__main__':# 创建应用程序实例app = QApplication([])# 创建主窗口实例window = MainWindow()# 显示主窗口window.show()# 运行应用程序事件循环app.exec_()

• 在上面例子中，
• 创建一个 QThreadPool 对象，用于管理线程池。
• 启动任务的方法 start_tasks，这个方法将在按钮点击时调用。循环创建并启动5个任务，每个任务的编号从0到4。使用线程池的 start 方法启动每个任务。

2.2.3 高级用法

• 控制最大线程数

  可以设置线程池中线程的最大数量，以避免系统过载。

self.thread_pool.setMaxThreadCount(10)

• 处理任务结果

  可以使用自定义信号或回调函数来处理任务完成后的结果。

from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget
from PyQt5.QtCore import pyqtSignal, QObject, QRunnable, QThreadPool, pyqtSlot# 定义一个信号类，用于任务之间的通信
class WorkerSignals(QObject):finished = pyqtSignal()  # 定义一个无参数的信号，表示任务完成result = pyqtSignal(object)  # 定义一个带参数的信号，用于传递任务结果# 定义一个任务类，继承自QRunnable
class Task(QRunnable):def __init__(self, n):super().__init__()self.n = n  # 初始化任务编号self.signals = WorkerSignals()  # 创建信号实例@pyqtSlot()def run(self):print(f"Task {self.n} is running")  # 输出任务开始的消息import timetime.sleep(2)  # 模拟长时间任务result = f"Result of task {self.n}"  # 生成任务结果self.signals.result.emit(result)  # 发出任务结果信号self.signals.finished.emit()  # 发出任务完成信号# 定义主窗口类，继承自QMainWindow
class MainWindow(QMainWindow):def __init__(self):super().__init__()self.initUI()  # 初始化用户界面def initUI(self):# 创建一个按钮，文本为"Start Tasks"self.button = QPushButton('Start Tasks', self)# 将按钮的点击信号连接到start_tasks方法self.button.clicked.connect(self.start_tasks)# 创建垂直布局，并将按钮添加到布局中layout = QVBoxLayout()layout.addWidget(self.button)# 创建一个容器窗口，将布局设置为容器的布局container = QWidget()container.setLayout(layout)# 将容器设置为主窗口的中心窗口self.setCentralWidget(container)# 创建一个线程池self.thread_pool = QThreadPool()# 启动任务的方法def start_tasks(self):for i in range(5):task = Task(i)  # 创建任务对象，并传递编号# 连接任务的结果信号到处理结果的方法task.signals.result.connect(self.handle_result)# 连接任务的完成信号到任务完成的方法task.signals.finished.connect(self.task_finished)# 启动任务self.thread_pool.start(task)# 处理任务结果的方法def handle_result(self, result):print(result)  # 输出任务结果# 任务完成的方法def task_finished(self):print("Task finished")  # 输出任务完成消息if __name__ == '__main__':# 创建应用程序实例app = QApplication([])# 创建主窗口实例window = MainWindow()# 显示主窗口window.show()# 运行应用程序事件循环app.exec_()

运行机制

• 主窗口初始化时创建一个按钮。
• 点击按钮时，启动 5 个任务，每个任务将在独立线程中运行。
• 每个任务通过信号将任务的运行和完成消息发送到主窗口。
• 主窗口接收到信号后，处理任务结果并输出消息。

2.2.4 总结

• QRunnable：创建可运行的任务对象，重载run()方法定义任务逻辑。
• QThreadPool：管理和执行QRunnable任务，提供线程池功能以提高多线程任务执行效率。
• 信号与槽：通过自定义信号来处理任务结果和任务完成事件，确保主线程能够正确响应多线程任务的状态。

2.3 实践

下面是一个简单的PyQt5应用程序，使用QThread在后台执行任务并更新UI。

import sys
from PyQt5.QtCore import QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget, QPushButton, QProgressBar# 定义一个继承自 QThread 的工作线程类
class WorkerThread(QThread):# 定义一个整型信号，用于发送进度信息progress = pyqtSignal(int)def __init__(self):super().__init__()# 线程运行的方法def run(self):# 模拟长时间任务，更新进度for i in range(101):self.msleep(50)  # 休眠 50 毫秒，模拟长时间任务self.progress.emit(i)  # 发射进度信号# 主窗口类，继承自 QMainWindow
class MainWindow(QMainWindow):def __init__(self):super().__init__()self.initUI()# 初始化用户界面def initUI(self):self.setWindowTitle("QThread Example")  # 设置窗口标题self.setGeometry(100, 100, 300, 150)  # 设置窗口位置和大小self.layout = QVBoxLayout()  # 创建垂直布局# 创建一个按钮，连接到 start_task 方法self.button = QPushButton("Start Task", self)self.button.clicked.connect(self.start_task)# 创建一个进度条，范围为 0 到 100self.progress_bar = QProgressBar(self)self.progress_bar.setRange(0, 100)# 将按钮和进度条添加到布局中self.layout.addWidget(self.button)self.layout.addWidget(self.progress_bar)container = QWidget()  # 创建一个容器窗口container.setLayout(self.layout)  # 将布局设置为容器的布局self.setCentralWidget(container)  # 将容器设置为主窗口的中心窗口self.show()  # 显示主窗口# 启动任务的方法def start_task(self):self.thread = WorkerThread()  # 创建工作线程对象self.thread.progress.connect(self.update_progress)  # 将线程的进度信号连接到更新进度的方法self.thread.start()  # 启动线程# 更新进度的槽函数@pyqtSlot(int)def update_progress(self, value):self.progress_bar.setValue(value)  # 设置进度条的值为接收到的进度值# 主函数
if __name__ == "__main__":app = QApplication(sys.argv)  # 创建应用程序实例window = MainWindow()  # 创建主窗口实例sys.exit(app.exec_())  # 运行应用程序事件循环，并返回退出状态码

代码解释：

• WorkerThread继承自QThread，并重载了run方法，模拟一个长时间任务。

• 使用progress信号来传递进度信息。

• MainWindow类继承自QMainWindow，并初始化UI，包括一个按钮和一个进度条。

• start_task方法创建并启动WorkerThread线程，连接progress信号到update_progress槽函数。

• update_progress槽函数更新进度条的值。

• 运行结果如下

  

3. 常见问题与调试技巧

3.1 线程安全

在开发多线程的应用程序时，确保线程安全是一个关键问题。线程安全是指多个线程可以正确且无冲突地访问和修改共享资源。以下是关于确保线程安全的一些常见问题和调试技巧：

3.1.1. 竞争条件

问题：竞争条件发生在两个或多个线程同时访问共享数据，并且至少有一个线程修改数据时。竞争条件可能导致数据不一致或崩溃。

解决方案：

• 使用锁：在访问共享数据时使用锁（例如threading.Lock）来确保一次只有一个线程可以访问数据。

示例

import threading# 定义一个计数器类
class Counter:def __init__(self):self.count = 0  # 初始化计数器self.lock = threading.Lock()  # 创建一个互斥锁，用于保护计数器的操作# 计数器增加方法，使用了互斥锁确保线程安全def increment(self):with self.lock:self.count += 1counter = Counter()  # 创建计数器实例# 定义一个工作函数，用于在多个线程中调用
def worker():for _ in range(1000):counter.increment()  # 每个线程调用计数器的增加方法# 创建10个线程，每个线程都调用 worker 函数
threads = [threading.Thread(target=worker) for _ in range(10)]
for thread in threads:thread.start()  # 启动线程
for thread in threads:thread.join()  # 等待所有线程结束print(counter.count)  # 输出计数器的值，期望输出为 10000

3.1.2. 死锁

问题：死锁发生在两个或多个线程相互等待对方释放资源时，导致它们都无法继续执行。

解决方案：

• 避免嵌套锁定：尽量避免嵌套锁定，即一个线程在持有一个锁的同时尝试获取另一个锁。
• 使用超时：使用超时来获取锁，如果超过时间限制则放弃，以避免死锁。
• 顺序锁定：确保所有线程以相同的顺序获取锁。

示例

import threading# 创建两个互斥锁对象
lock1 = threading.Lock()
lock2 = threading.Lock()# 定义 worker1 函数
def worker1():with lock1:  # 获取 lock1 锁print("Worker1 acquired lock1")with lock2:  # 获取 lock2 锁print("Worker1 acquired lock2")# 定义 worker2 函数
def worker2():with lock2:  # 获取 lock2 锁print("Worker2 acquired lock2")with lock1:  # 获取 lock1 锁print("Worker2 acquired lock1")# 创建两个线程，分别执行 worker1 和 worker2 函数
thread1 = threading.Thread(target=worker1)
thread2 = threading.Thread(target=worker2)# 启动线程
thread1.start()
thread2.start()# 等待线程结束
thread1.join()
thread2.join()

1. worker1 函数：
   • 在 worker1 函数中，首先获取了 lock1 锁，然后再获取 lock2 锁。
   • 这种嵌套锁的方式可能会导致死锁，因为在后续的 worker2 函数中，线程可能先获取 lock2 锁，再获取 lock1 锁，与 worker1 中的获取顺序相反，从而造成死锁。
2. worker2 函数：
   • 在 worker2 函数中，首先获取了 lock2 锁，然后再获取 lock1 锁。
   • 这样的获取顺序与 worker1 函数中的相反，可能导致死锁。

3.2. 调试技巧

• 问题：多线程的程序更难调试，因为它们的行为可能不确定，且错误可能是间歇性的。

• 解决方案：

  • 日志记录：使用日志记录（例如logging模块）来跟踪线程的行为和状态。

  • 调试器：使用调试器（如pdb）来单步执行和检查线程的状态。

  • 测试用例：编写单元测试和多线程测试用例来验证代码的正确性。

4. 参考资料

• 官方文档：深入阅读PyQt5的官方文档和相关示例。