如何在执行中途终止wxPython中的CountingThread线程类？

现有wxPython程序中，点击「Start Counter」按钮会启动非阻塞的CountingThread线程，每隔2秒将数值递增3直至达到12。需要实现「Stop Counter」按钮，使其能在任意阶段终止该线程。

修改思路 #

Python不支持强制终止线程（存在安全风险），最优方案是使用线程安全的事件标志让线程主动退出循环，完成终止。

具体修改步骤 #

1. 在CountingThread类中添加线程停止标志self._stop_event，基于threading.Event()实现
2. 修改run方法的循环条件，每次循环前检测停止标志，若标志被设置则退出循环
3. 给CountingThread添加stop方法，用于触发停止标志
4. 在CountingFrame中保存线程实例，避免线程对象被局部销毁，确保Stop按钮能调用到目标线程的停止方法

修改后的完整代码 #

#-----------------------------------------------------------------------------#
import wx
import time
import threading
number = 0
#-----------------------------------------------------------------------------#

class CountingFrame(wx.Frame):      
    def __init__(self, parent):
        wx.Frame.__init__(self, parent)

        # Attributes
        self._counter = wx.StaticText(self, label="0")
        self._count_thread = None  # 保存线程实例

        # Layout
        sizer = wx.BoxSizer(wx.VERTICAL)
        start_button = wx.Button(self, wx.ID_ANY, "Start Counter")
        stop_button = wx.Button(self, wx.ID_ANY, "Stop Counter")
        sizer.AddMany([(start_button, 0, wx.ALIGN_CENTER),
                       (stop_button, 0, wx.ALIGN_CENTER),
                       ((15, 15), 0),
                       (self._counter, 0, wx.ALIGN_CENTER)])

        self.SetSizer(sizer)
        self.SetMinSize((300, 300))        
        
        # Event Handlers
        self.Bind(wx.EVT_BUTTON, self.OnStart, start_button)
        self.Bind(wx.EVT_BUTTON, self.OnStop, stop_button)
        self.Bind(EVT_COUNT, self.OnCount)

    def OnStart(self, evt):
        # 避免重复启动线程
        if self._count_thread is None or not self._count_thread.is_alive():
            self._count_thread = CountingThread(self)
            self._count_thread.start()
        
    def OnStop(self, evt):
        # 检查线程是否存在且存活，调用停止方法
        if self._count_thread is not None and self._count_thread.is_alive():
            self._count_thread.stop()

    def OnCount(self, evt):     
        val = str(int(self._counter.GetLabel()) + evt.GetValue())
        self._counter.SetLabel(val)

#-----------------------------------------------------------------------------#

## Runs on its own thread and calls CountEvent to increment num    
class CountingThread(threading.Thread):        
    def __init__(self, parent):
        threading.Thread.__init__(self)
        self._parent = parent
        self._stop_event = threading.Event()  # 线程停止标志

    def stop(self):
        # 设置停止标志
        self._stop_event.set()

    def run(self):
        limit = 5
        increment_amt = 1
        # 循环条件加入停止标志检测
        while increment_amt < limit and not self._stop_event.is_set():
            time.sleep(2) # 模拟计算时间
            # 若在sleep后标志被设置，直接退出，避免发送多余事件
            if self._stop_event.is_set():
                break
            evt = CountEvent(myEVT_COUNT)  
            wx.PostEvent(self._parent, evt)
            increment_amt +=1
    
#-----------------------------------------------------------------------------#
    
## Performs the event: increments num by 3
myEVT_COUNT = wx.NewEventType()
EVT_COUNT = wx.PyEventBinder(myEVT_COUNT)
class CountEvent(wx.PyCommandEvent): 
    """Event to signal that a count value is ready"""
    def __init__(self, eid):
        wx.PyCommandEvent.__init__(self, eid)

    def GetValue(self):
        global number
        temp = number + 3
        number = temp  # 修正原代码：更新全局变量number的值
        return temp

#-----------------------------------------------------------------------------#

if __name__ == '__main__':
    APP = wx.App(False)
    FRAME = CountingFrame(None)
    FRAME.Show()
    APP.MainLoop()

#-----------------------------------------------------------------------------#

额外说明 #

• 原代码中CountEvent.GetValue()只计算了增量但未更新全局变量number，已修正该问题
• 添加了重复启动线程的判断，避免多次点击Start按钮创建多个线程
• 在run方法的sleep后再次检测停止标志，避免线程在sleep期间被停止后仍发送事件

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