Python与PyQt5并行处理函数执行结果重叠问题求助

核心问题 #

1. 信号共享导致结果混乱：thread_X_1、thread_X_2的signal是类属性，所有线程实例共用同一个信号，因此所有子线程的结果会被所有监听者接收，无法区分归属。
2. 循环引用逻辑错误：thread_X_1和thread_X_2的__init__里创建Window，而Window又会创建这两个子线程，导致无限递归创建对象，完全破坏线程逻辑。
3. 线程执行逻辑错误：thread_X的子线程启动逻辑写在__init__中，而__init__是在主线程执行的，没有利用QThread的run()方法实现真正的并行。

修复后的代码示例 #

import sys
import numpy as np
from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtWidgets import QMainWindow, QApplication

class thread_X_1(QThread):
    # 增加标识参数，区分属于哪个父任务
    signal = pyqtSignal(np.ndarray, str)
    def __init__(self, task_id):
        super().__init__()
        self.task_id = task_id  # 记录所属的父任务ID

    def run(self):
        # 替换为实际子任务逻辑，示例返回随机数组
        result = np.random.rand(2,2)
        self.signal.emit(result, self.task_id)

class thread_X_2(QThread):
    signal = pyqtSignal(np.ndarray, str)
    def __init__(self, task_id):
        super().__init__()
        self.task_id = task_id

    def run(self):
        result = np.random.rand(3,3)
        self.signal.emit(result, self.task_id)

class TaskContainer:
    # 重命名避免与UI窗口混淆，负责管理单个父任务的子线程
    def __init__(self, task_id):
        self.task_id = task_id
        self.results = []  # 收集当前任务的子线程结果
        self.pX_1 = thread_X_1(task_id)
        self.pX_2 = thread_X_2(task_id)

        # 绑定子线程信号到当前容器的处理函数
        self.pX_1.signal.connect(self.collect_result)
        self.pX_2.signal.connect(self.collect_result)

        self.pX_1.start()
        self.pX_2.start()

        self.pX_1.wait()
        self.pX_2.wait()

    def collect_result(self, result, task_id):
        # 只收集属于当前任务的结果
        if task_id == self.task_id:
            self.results.append(result)

class thread_X(QThread):
    # 返回当前任务的所有子结果和任务ID
    signal = pyqtSignal(list, str)
    def __init__(self, task_id):
        super().__init__()
        self.task_id = task_id

    def run(self):
        # 子任务逻辑放到run()里，才会在新线程执行
        container = TaskContainer(self.task_id)
        self.signal.emit(container.results, self.task_id)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.task_results = {}  # 按任务ID存储结果

        # 创建两个并行任务，赋予唯一ID区分
        self.p1 = thread_X("task_1")
        self.p1.signal.connect(self.handle_task_result)
        self.p1.start()

        self.p2 = thread_X("task_2")
        self.p2.signal.connect(self.handle_task_result)
        self.p2.start()

    def handle_task_result(self, sub_results, task_id):
        self.task_results[task_id] = sub_results
        print(f"任务{task_id}完成，子结果数量：{len(sub_results)}")
        # 此处可添加结果展示或后续处理逻辑

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    sys.exit(app.exec_())

关键修复点 #

• 移除循环引用：删除thread_X_1、thread_X_2中创建Window的代码，彻底解决无限递归问题。
• 信号增加标识：子线程信号传递任务ID，确保结果能正确归属到对应的父任务。
• 线程逻辑移到run()：thread_X的子任务启动逻辑放到run()方法中，真正实现并行执行。
• 结果收集隔离：每个TaskContainer只收集自身所属父任务的子线程结果，再通过thread_X的信号返回给主线程，避免结果混淆。

内容的提问来源于stack exchange，提问作者RACHID BEN ABDELMALEK