You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.16 23:45:39