PyQt5多线程控制X移动程序触发SIGSEGV崩溃求助
问题
我正在开发一款可在指定框架内移动目标的程序,需求为按下按键启动while循环,释放按键时停止循环。采用多线程方案实现,但程序运行一段时间后会出现各类报错或完全冻结,且频繁被signal 11 SIGSEGV信号中断。
开发环境:MacBook(OS 12.6)+PyCharm
原代码如下:
# GLOBAL import faulthandler; faulthandler.enable() import sys import time import requests from pynput import keyboard from pynput.keyboard import Key from qtpy import QtWidgets from PyQt5 import QtCore from ui.mainwindow import Ui_MainWindow # Globale Variablen: PAN = 250 TILT = 150 SPEED = 1 TIME = 0.01 KEYS = [] THREADS = 0 DIRECTION = [] # HTTP SECTION payload = {"name": "MK"} r = requests.get("https://httpbin.org/get", params=payload) print(r.status_code) # BUILD GUI app = QtWidgets.QApplication(sys.argv) class MainWindow(QtWidgets.QMainWindow): def __init__(self, parent=None): super().__init__(parent) self.ui = Ui_MainWindow() # Laden des Style Frames self.ui.setupUi(self) # Funktion aus Ui_MainWindow ausführen (Fenster zeichnen) self.ui.tilt.setText(str(TILT)) self.ui.pan.setText(str(PAN)) self.ui.speed.setText(str(SPEED)) ## Button Funktionsaufrufe: self.thread = {} self.ui.btnUp.clicked.connect(self.moveUP) self.ui.btnDown.clicked.connect(self.moveDOWN) self.ui.btnLeft.clicked.connect(self.moveLEFT) self.ui.btnRight.clicked.connect(self.moveRIGHT) self.ui.btnCenter.clicked.connect(self.stop_worker) self.ui.speed.textChanged.connect(self.setSpeed) # Funktion zum bewegen über die Buttons def startUp(self): self.thread[1] = ThreadClass(parent=None, index=1) self.thread[1].start() self.thread[1].any_signal.connect(self.my_function) time.sleep(0.5) self.thread[1].stop() print('init thread 1 on startup') self.thread[2] = ThreadClass(parent=None, index=2) self.thread[2].start() time.sleep(0.5) self.thread[2].stop() print('init thread 2 on startup') self.thread[3] = ThreadClass(parent=None, index=3) self.thread[3].start() time.sleep(0.5) self.thread[3].stop() print('init thread 3 on startup') self.thread[4] = ThreadClass(parent=None, index=4) self.thread[4].start() time.sleep(0.5) self.thread[4].stop() print('init thread 4 on startup') print('... \n \n \n') print('Status = Ready:') print('Keys = ', KEYS) def moveUP(self): if bool(self.thread[2].isRunning()): self.stopDOWN() self.thread[1] = ThreadClass(parent=None, index=1) self.thread[1].start() self.thread[1].any_signal.connect(self.my_function) #self.ui.btnUp.setEnabled(False) def moveDOWN(self): # global DIRECTION # DIRECTION.append('Down') self.thread[2] = ThreadClass(parent=None, index=2) self.thread[2].start() self.thread[2].any_signal.connect(self.my_function) #self.ui.btnDown.setEnabled(False) def moveLEFT(self): # global DIRECTION # DIRECTION.append('Left') if bool(self.thread[4].isRunning()): self.stopRIGHT() self.thread[3] = ThreadClass(parent=None, index=3) self.thread[3].start() self.thread[3].any_signal.connect(self.my_function) #self.ui.btnLeft.setEnabled(False) def moveRIGHT(self): # global DIRECTION # DIRECTION.append('Right') self.thread[4] = ThreadClass(parent=None, index=4) self.thread[4].start() self.thread[4].any_signal.connect(self.my_function) #self.ui.btnRight.setEnabled(False) def stopUP(self): global KEYS if bool(self.thread[1].isRunning()): self.thread[1].stop() #self.ui.btnUp.setEnabled(True) KEYS.remove(Key.up) # DIRECTION.remove('Up') def stopDOWN(self): global KEYS if bool(self.thread[2].isRunning()): self.thread[2].stop() #self.ui.btnDown.setEnabled(True) KEYS.remove(Key.down) # DIRECTION.remove('Down') def stopLEFT(self): global KEYS if bool(self.thread[3].isRunning()): self.thread[3].stop() #self.ui.btnLeft.setEnabled(True) KEYS.remove(Key.left) # DIRECTION.remove('Left') def stopRIGHT(self): global KEYS if bool(self.thread[4].isRunning()): self.thread[4].stop() #self.ui.btnRight.setEnabled(True) KEYS.remove(Key.right) # DIRECTION.remove('Right') def stop_worker(self): global DIRECTION if 'Up' in DIRECTION: self.thread[1].stop() #self.ui.btnUp.setEnabled(True) DIRECTION.remove('Up') if 'Down' in DIRECTION: self.thread[2].stop() #self.ui.btnDown.setEnabled(True) DIRECTION.remove('Down') if 'Left' in DIRECTION: self.thread[3].stop() #self.ui.btnLeft.setEnabled(True) if 'Right' in DIRECTION: self.thread[4].stop() #self.ui.btnRight.setEnabled(True) DIRECTION.remove('Right') def my_function(self, counter): global KEYS cnt = counter index = self.sender().index if index == 1: print(cnt) def setSpeed(self): global SPEED if self.ui.speed.text(): SPEED = int(self.ui.speed.text()) # Funktion zum bewegen über die Tasten def on_press(key): global KEYS if key not in KEYS: if key == Key.up: MainWindow.moveUP(window) KEYS.append(key) print('added: ' + str(key)) if key == Key.down: if Key.up in KEYS: pass else: MainWindow.moveDOWN(window) KEYS.append(key) print('added: ' + str(key)) if key == Key.left: MainWindow.moveLEFT(window) KEYS.append(key) print('added: ' + str(key)) if key == Key.right: if Key.left in KEYS: pass else: MainWindow.moveRIGHT(window) KEYS.append(key) print('added: ' + str(key)) def on_release(key): global KEYS if key in KEYS: if key == Key.up and Key.up in KEYS: MainWindow.stopUP(window) if key == Key.down and Key.down in KEYS: MainWindow.stopDOWN(window) if key == Key.left and Key.left in KEYS: MainWindow.stopLEFT(window) if key == Key.right and Key.right in KEYS: MainWindow.stopRIGHT(window) print(KEYS) if key == Key.esc: MainWindow.stop_worker(window) # Collect events until released listener = keyboard.Listener( on_press=on_press, on_release=on_release) listener.start() class ThreadClass(QtCore.QThread): any_signal = QtCore.pyqtSignal(int) def __init__(self, parent=None, index=0): super(ThreadClass, self).__init__(parent) self.index = index self.is_running = True def run(self): global TILT global PAN global TIME global SPEED print('Starting thread...', self.index) i = self.index while (True): if i == 1: if TILT <= 0: pass else: TILT -= SPEED window.ui.tilt.setText(str(TILT)) window.ui.target.setGeometry(PAN, TILT, 10, 16) # print("T", TILT) time.sleep(TIME) if i == 2: if TILT >= 301: pass else: TILT += SPEED window.ui.tilt.setText(str(TILT)) window.ui.target.setGeometry(PAN, TILT, 10, 16) # print('T', TILT) time.sleep(TIME) if i == 3: if PAN <= 0: pass else: PAN -= SPEED window.ui.pan.setText(str(PAN)) window.ui.target.setGeometry(PAN, TILT, 10, 16) # print("P", PAN) time.sleep(TIME) if i == 4: if PAN >= 500: pass else: PAN += SPEED window.ui.pan.setText(str(PAN)) window.ui.target.setGeometry(PAN, TILT, 10, 16) # print("P", PAN) time.sleep(TIME) def stop(self): self.is_running = False print('Stopping thread...', self.index) self.terminate() window = MainWindow() window.show() window.startUp() sys.exit(app.exec_())
问题分析与修复方案
- 禁止子线程直接操作UI:Qt的UI组件只能在主线程更新,原代码中子线程直接修改UI元素是导致SIGSEGV崩溃的核心原因。修复后通过
pos_updated信号将位置数据传递到主线程,由主线程的update_ui方法统一更新UI。 - 替换危险的terminate()方法:
self.terminate()会强制终止线程,引发资源泄漏和内存错误。修复后通过is_running标志位控制循环自然退出,线程会正常执行完逻辑后结束。 - 线程安全的全局变量访问:原代码多线程直接读写全局
PAN、TILT存在数据竞争。修复后每个线程初始化时获取全局值,运行中维护自身副本,更新后通过信号同步到全局变量。 - 键盘回调的主线程调用:pynput监听器运行在独立线程,直接调用UI方法会引发线程安全问题。修复后用
QtCore.QMetaObject.invokeMethod确保所有UI操作在主线程执行。 - 移除主线程阻塞操作:原
startUp方法中的time.sleep会阻塞主线程导致UI卡顿,直接移除该逻辑,线程无需强制启动再停止。
修复后的代码:
# GLOBAL import faulthandler; faulthandler.enable() import sys import time import requests from pynput import keyboard from pynput.keyboard import Key from qtpy import QtWidgets from PyQt5 import QtCore from ui.mainwindow import Ui_MainWindow # Globale Variablen: PAN = 250 TILT = 150 SPEED = 1 TIME = 0.01 KEYS = [] # HTTP SECTION payload = {"name": "MK"} r = requests.get("https://httpbin.org/get", params=payload) print(r.status_code) # BUILD GUI app = QtWidgets.QApplication(sys.argv) class MainWindow(QtWidgets.QMainWindow): def __init__(self, parent=None): super().__init__(parent) self.ui = Ui_MainWindow() self.ui.setupUi(self) self.ui.tilt.setText(str(TILT)) self.ui.pan.setText(str(PAN)) self.ui.speed.setText(str(SPEED)) self.thread = {} self.ui.btnUp.clicked.connect(self.moveUP) self.ui.btnDown.clicked.connect(self.moveDOWN) self.ui.btnLeft.clicked.connect(self.moveLEFT) self.ui.btnRight.clicked.connect(self.moveRIGHT) self.ui.btnCenter.clicked.connect(self.stop_worker) self.ui.speed.textChanged.connect(self.setSpeed) def moveUP(self): if self.thread.get(2) and self.thread[2].isRunning(): self.stopDOWN() self.thread[1] = ThreadClass(parent=self, index=1, speed=SPEED, time=TIME) self.thread[1].pos_updated.connect(self.update_ui) self.thread[1].start() def moveDOWN(self): if self.thread.get(1) and self.thread[1].isRunning(): self.stopUP() self.thread[2] = ThreadClass(parent=self, index=2, speed=SPEED, time=TIME) self.thread[2].pos_updated.connect(self.update_ui) self.thread[2].start() def moveLEFT(self): if self.thread.get(4) and self.thread[4].isRunning(): self.stopRIGHT() self.thread[3] = ThreadClass(parent=self, index=3, speed=SPEED, time=TIME) self.thread[3].pos_updated.connect(self.update_ui) self.thread[3].start() def moveRIGHT(self): if self.thread.get(3) and self.thread[3].isRunning(): self.stopLEFT() self.thread[4] = ThreadClass(parent=self, index=4, speed=SPEED, time=TIME) self.thread[4].pos_updated.connect(self.update_ui) self.thread[4].start() def stopUP(self): global KEYS if self.thread.get(1) and self.thread[1].isRunning(): self.thread[1].stop() KEYS.remove(Key.up) def stopDOWN(self): global KEYS if self.thread.get(2) and self.thread[2].isRunning(): self.thread[2].stop() KEYS.remove(Key.down) def stopLEFT(self): global KEYS if self.thread.get(3) and self.thread[3].isRunning(): self.thread[3].stop() KEYS.remove(Key.left) def stopRIGHT(self): global KEYS if self.thread.get(4) and self.thread[4].isRunning(): self.thread[4].stop() KEYS.remove(Key.right) def stop_worker(self): for idx in self.thread: if self.thread[idx].isRunning(): self.thread[idx].stop() global KEYS KEYS.clear() def setSpeed(self): global SPEED if self.ui.speed.text(): try: SPEED = int(self.ui.speed.text()) except ValueError: SPEED = 1 def update_ui(self, pan, tilt): global PAN, TILT PAN = pan TILT = tilt self.ui.pan.setText(str(PAN)) self.ui.tilt.setText(str(TILT)) self.ui.target.setGeometry(PAN, TILT, 10, 16) # 键盘事件处理 def on_press(key): global KEYS if key not in KEYS: if key == Key.up: QtCore.QMetaObject.invokeMethod(window, "moveUP") KEYS.append(key) print('added: ' + str(key)) if key == Key.down: if Key.up not in KEYS: QtCore.QMetaObject.invokeMethod(window, "moveDOWN") KEYS.append(key) print('added: ' + str(key)) if key == Key.left: QtCore.QMetaObject.invokeMethod(window, "moveLEFT") KEYS.append(key) print('added: ' + str(key)) if key == Key.right: if Key.left not in KEYS: QtCore.QMetaObject.invokeMethod(window, "moveRIGHT") KEYS.append(key) print('added: ' + str(key)) def on_release(key): global KEYS if key in KEYS: if key == Key.up: QtCore.QMetaObject.invokeMethod(window, "stopUP") if key == Key.down: QtCore.QMetaObject.invokeMethod(window, "stopDOWN") if key == Key.left: QtCore.Q
相关产品推荐
相关产品推荐

