如何实现双线程信号均发射时触发调用父类指定函数

针对你持续运行的摄像头UI+位置算法场景，核心思路是用状态跟踪+条件检查的方式，在两个线程的信号都收到时触发目标函数，完全不需要依赖join()（适配主循环持续运行的场景）。

核心实现逻辑 #

1. 定义状态变量：用两个布尔标记跟踪每个线程的计算结果是否就绪
2. 信号回调：每个线程完成计算后，更新对应状态变量并触发条件检查
3. 条件触发：当两个状态标记都为True时，调用目标处理函数，随后重置状态准备下一轮信号

通用Python threading示例 #

import threading
import time

class CameraProcessor:
    def __init__(self):
        # 状态跟踪标记：记录两个摄像头的计算结果是否就绪
        self._cam1_ready = False
        self._cam2_ready = False
        # 存储线程计算出的位置结果
        self.cam1_pos = None
        self.cam2_pos = None
        # 线程安全锁：避免多线程同时修改状态导致的竞态错误
        self._state_lock = threading.Lock()

        # 启动两个持续运行的摄像头处理线程
        self.cam_thread1 = threading.Thread(target=self._cam1_process_loop, daemon=True)
        self.cam_thread2 = threading.Thread(target=self._cam2_process_loop, daemon=True)
        self.cam_thread1.start()
        self.cam_thread2.start()

    def _cam1_process_loop(self):
        while True:
            # 模拟摄像头帧捕获+位置计算逻辑
            self.cam1_pos = self._calculate_position_from_cam1()
            # 发送信号（调用回调更新状态）
            self._on_cam1_signal()
            # 模拟30fps的帧间隔
            time.sleep(1/30)

    def _cam2_process_loop(self):
        while True:
            self.cam2_pos = self._calculate_position_from_cam2()
            self._on_cam2_signal()
            time.sleep(1/30)

    def _on_cam1_signal(self):
        with self._state_lock:
            self._cam1_ready = True
            self._check_trigger_condition()

    def _on_cam2_signal(self):
        with self._state_lock:
            self._cam2_ready = True
            self._check_trigger_condition()

    def _check_trigger_condition(self):
        # 两个信号均已收到时，触发目标处理函数
        if self._cam1_ready and self._cam2_ready:
            self._update_ui_and_run_location_algorithm(self.cam1_pos, self.cam2_pos)
            # 重置状态，等待下一轮信号
            self._cam1_ready = False
            self._cam2_ready = False

    # ---------------------- 业务逻辑占位 ----------------------
    def _calculate_position_from_cam1(self):
        # 实际替换为摄像头帧处理、位置计算代码
        return 100  # 模拟返回的整数信号

    def _calculate_position_from_cam2(self):
        return 200  # 模拟返回的整数信号

    def _update_ui_and_run_location_algorithm(self, pos1, pos2):
        # 你的目标函数：更新UI画面、执行位置查找逻辑
        print(f"更新UI：摄像头1位置{pos1}，摄像头2位置{pos2}，执行位置匹配")

# 主循环
if __name__ == "__main__":
    processor = CameraProcessor()
    while True:
        # 主循环可处理其他UI事件或后台逻辑
        time.sleep(0.1)

Qt框架适配示例（针对UI场景） #

如果用Qt开发UI，直接结合信号槽机制实现即可：

from PyQt5.QtCore import QThread, pyqtSignal, QObject
from PyQt5.QtWidgets import QApplication
import sys

class CameraWorker(QThread):
    # 定义线程完成计算后的信号
    result_ready = pyqtSignal(int)

    def __init__(self, cam_id):
        super().__init__()
        self.cam_id = cam_id

    def run(self):
        while True:
            # 模拟帧处理+位置计算
            pos = self._compute_position()
            self.result_ready.emit(pos)
            self.msleep(33)  # 约30fps的帧间隔

    def _compute_position(self):
        return self.cam_id * 100  # 模拟整数结果

class MainProcessor(QObject):
    def __init__(self):
        super().__init__()
        self._cam1_pos = None
        self._cam2_pos = None
        self._cam1_ready = False
        self._cam2_ready = False

        # 初始化并启动摄像头线程
        self.worker1 = CameraWorker(1)
        self.worker2 = CameraWorker(2)

        # 连接信号到回调函数
        self.worker1.result_ready.connect(self._on_cam1_result)
        self.worker2.result_ready.connect(self._on_cam2_result)

        self.worker1.start()
        self.worker2.start()

    def _on_cam1_result(self, pos):
        self._cam1_pos = pos
        self._cam1_ready = True
        self._check_trigger()

    def _on_cam2_result(self, pos):
        self._cam2_pos = pos
        self._cam2_ready = True
        self._check_trigger()

    def _check_trigger(self):
        if self._cam1_ready and self._cam2_ready:
            self._update_ui_and_locate()
            self._cam1_ready = False
            self._cam2_ready = False

    def _update_ui_and_locate(self):
        # 更新UI画面、执行位置查找逻辑
        print(f"Qt UI更新：摄像头1位置{self._cam1_pos}，摄像头2位置{self._cam2_pos}")

if __name__ == "__main__":
    app = QApplication(sys.argv)
    processor = MainProcessor()
    sys.exit(app.exec_())

关键注意事项 #

• 线程安全：状态变量会被多线程同时修改，必须加锁（如第一个示例中的threading.Lock()），避免逻辑混乱
• 状态重置：每次触发目标函数后必须重置状态标记，否则会因线程持续发送信号导致重复触发
• 守护线程：将工作线程设为守护线程（daemon=True），确保主程序退出时线程自动终止，避免残留进程

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