wx.Python 杀死子线程
wx.Python killing children thread
我有一个 GUI 应用程序,运行 在我的主线程中,一些子线程在后台执行操作。我将关闭按钮绑定到一个方法,在关闭软件之前做一些事情,但我不知道如何停止子线程
self.Bind(wx.EVT_CLOSE, self.onCloseFrame)
def onCloseFrame(self, event):
do_stuff()
# stop threads here
self.Destroy()
有办法吗?
这是我模拟的测试。
它比需要的要复杂一些,因为我想要多个进度指示器,但您应该能够了解它的要点。
包括一个终止函数并触发它,如果你想在它自然结束之前结束线程。
在这种情况下,线程会在一分钟后超时,但它可能正在执行任何操作。
单击 Start
可同时拥有 运行 个线程。
import time
import wx
from threading import Thread
import wx.lib.newevent
progress_event, EVT_PROGRESS_EVENT = wx.lib.newevent.NewEvent()
class ProcessingFrame(wx.Frame):
def __init__(self, title, parent=None):
wx.Frame.__init__(self, parent=parent, title=title)
panel = wx.Panel(self)
self.parent = parent
self.btn = wx.Button(panel,label='Stop processing', size=(200,30), pos=(10,10))
self.btn.Bind(wx.EVT_BUTTON, self.OnExit)
self.progress = wx.Gauge(panel,size=(240,10), pos=(10,50), range=60)
self.process = wx.TextCtrl(panel,size = (200,250), pos=(10,100), style = wx.TE_MULTILINE)
#Bind to the progress event issued by the thread
self.Bind(EVT_PROGRESS_EVENT, self.OnProgress)
#Bind to Exit on frame close
self.Bind(wx.EVT_CLOSE, self.OnExit)
self.Show()
self.mythread = TestThread(self)
def OnProgress(self, event):
self.progress.SetValue(event.count)
self.process.write("Checking process: "+event.process+"\n")
self.Refresh()
if event.count >= 60:
self.OnExit(None)
def OnExit(self, event):
if self.mythread.isAlive():
self.mythread.terminate() # Shutdown the thread
self.mythread.join() # Wait for it to finish
self.Destroy()
class TestThread(Thread):
def __init__(self,parent_target):
Thread.__init__(self)
self.parent = parent_target
self.stopthread = False
self.time = time.time()
self.process = 1 # Testing only - mock process id
self.start() # start the thread
def run(self):
# A loop that will run for a while then terminate
while self.stopthread == False:
curr_loop = int(time.time() - self.time)
self.process += 10 # Testing only - mock process id
if curr_loop <= 60: # Update progress bar
time.sleep(0.1)
evt = progress_event(count=curr_loop,process=str(self.process))
#Send back current count for the progress bar
try:
wx.PostEvent(self.parent, evt)
except: # The parent frame has probably been destroyed
self.terminate()
def terminate(self):
self.stopthread = True
class MyPanel(wx.Panel):
def __init__(self, parent):
wx.Panel.__init__(self, parent)
self.text_count = 0
self.parent=parent
btn_start = wx.Button(self, wx.ID_ANY, label='Start Long running process', size=(180,30), pos=(10,10))
btn_test = wx.Button(self, wx.ID_ANY, label='Is the GUI still active?', size=(180,30), pos=(10,50))
self.txt = wx.TextCtrl(self, wx.ID_ANY, style= wx.TE_MULTILINE, pos=(10,90),size=(400,100))
btn_end = wx.Button(self, wx.ID_ANY, label='Stop All', size=(100,30), pos=(10,200))
btn_start.Bind(wx.EVT_BUTTON, self.Start_Process)
btn_test.Bind(wx.EVT_BUTTON, self.ActiveText)
btn_end.Bind(wx.EVT_BUTTON, self.End_AllProcesses)
self.activeframes = [] # 1 per live process
def Start_Process(self, event):
#thread_count = active_count()
thread_count = len(self.activeframes) + 1
self.activeframes.append(ProcessingFrame(title='Threaded Task '+str(thread_count), parent=self))
def End_AllProcesses(self, event):
while len(self.activeframes) > 0:
try:
self.activeframes[-1].OnExit(None)
self.activeframes.pop(-1)
except: # process may have finished or been stopped individually
self.activeframes.pop(-1)
def ActiveText(self,event):
self.text_count += 1
txt = "Gui is still active " + str(self.text_count)+"\n"
self.txt.write(txt)
class MainFrame(wx.Frame):
def __init__(self):
wx.Frame.__init__(self, None, title='Main Frame', size=(600,400))
panel = MyPanel(self)
self.Show()
if __name__ == '__main__':
app = wx.App(False)
frame = MainFrame()
app.MainLoop()
我有一个 GUI 应用程序,运行 在我的主线程中,一些子线程在后台执行操作。我将关闭按钮绑定到一个方法,在关闭软件之前做一些事情,但我不知道如何停止子线程
self.Bind(wx.EVT_CLOSE, self.onCloseFrame)
def onCloseFrame(self, event):
do_stuff()
# stop threads here
self.Destroy()
有办法吗?
这是我模拟的测试。
它比需要的要复杂一些,因为我想要多个进度指示器,但您应该能够了解它的要点。
包括一个终止函数并触发它,如果你想在它自然结束之前结束线程。
在这种情况下,线程会在一分钟后超时,但它可能正在执行任何操作。
单击 Start
可同时拥有 运行 个线程。
import time
import wx
from threading import Thread
import wx.lib.newevent
progress_event, EVT_PROGRESS_EVENT = wx.lib.newevent.NewEvent()
class ProcessingFrame(wx.Frame):
def __init__(self, title, parent=None):
wx.Frame.__init__(self, parent=parent, title=title)
panel = wx.Panel(self)
self.parent = parent
self.btn = wx.Button(panel,label='Stop processing', size=(200,30), pos=(10,10))
self.btn.Bind(wx.EVT_BUTTON, self.OnExit)
self.progress = wx.Gauge(panel,size=(240,10), pos=(10,50), range=60)
self.process = wx.TextCtrl(panel,size = (200,250), pos=(10,100), style = wx.TE_MULTILINE)
#Bind to the progress event issued by the thread
self.Bind(EVT_PROGRESS_EVENT, self.OnProgress)
#Bind to Exit on frame close
self.Bind(wx.EVT_CLOSE, self.OnExit)
self.Show()
self.mythread = TestThread(self)
def OnProgress(self, event):
self.progress.SetValue(event.count)
self.process.write("Checking process: "+event.process+"\n")
self.Refresh()
if event.count >= 60:
self.OnExit(None)
def OnExit(self, event):
if self.mythread.isAlive():
self.mythread.terminate() # Shutdown the thread
self.mythread.join() # Wait for it to finish
self.Destroy()
class TestThread(Thread):
def __init__(self,parent_target):
Thread.__init__(self)
self.parent = parent_target
self.stopthread = False
self.time = time.time()
self.process = 1 # Testing only - mock process id
self.start() # start the thread
def run(self):
# A loop that will run for a while then terminate
while self.stopthread == False:
curr_loop = int(time.time() - self.time)
self.process += 10 # Testing only - mock process id
if curr_loop <= 60: # Update progress bar
time.sleep(0.1)
evt = progress_event(count=curr_loop,process=str(self.process))
#Send back current count for the progress bar
try:
wx.PostEvent(self.parent, evt)
except: # The parent frame has probably been destroyed
self.terminate()
def terminate(self):
self.stopthread = True
class MyPanel(wx.Panel):
def __init__(self, parent):
wx.Panel.__init__(self, parent)
self.text_count = 0
self.parent=parent
btn_start = wx.Button(self, wx.ID_ANY, label='Start Long running process', size=(180,30), pos=(10,10))
btn_test = wx.Button(self, wx.ID_ANY, label='Is the GUI still active?', size=(180,30), pos=(10,50))
self.txt = wx.TextCtrl(self, wx.ID_ANY, style= wx.TE_MULTILINE, pos=(10,90),size=(400,100))
btn_end = wx.Button(self, wx.ID_ANY, label='Stop All', size=(100,30), pos=(10,200))
btn_start.Bind(wx.EVT_BUTTON, self.Start_Process)
btn_test.Bind(wx.EVT_BUTTON, self.ActiveText)
btn_end.Bind(wx.EVT_BUTTON, self.End_AllProcesses)
self.activeframes = [] # 1 per live process
def Start_Process(self, event):
#thread_count = active_count()
thread_count = len(self.activeframes) + 1
self.activeframes.append(ProcessingFrame(title='Threaded Task '+str(thread_count), parent=self))
def End_AllProcesses(self, event):
while len(self.activeframes) > 0:
try:
self.activeframes[-1].OnExit(None)
self.activeframes.pop(-1)
except: # process may have finished or been stopped individually
self.activeframes.pop(-1)
def ActiveText(self,event):
self.text_count += 1
txt = "Gui is still active " + str(self.text_count)+"\n"
self.txt.write(txt)
class MainFrame(wx.Frame):
def __init__(self):
wx.Frame.__init__(self, None, title='Main Frame', size=(600,400))
panel = MyPanel(self)
self.Show()
if __name__ == '__main__':
app = wx.App(False)
frame = MainFrame()
app.MainLoop()